commit d0db98b690d548ea4b4384ed18755a1e71c0ddb8 Author: Louis Date: Wed Aug 5 11:22:47 2026 +0200 feat: add scoped NetBox ZIP export and import plugin diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2615e23 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +*.egg-info/ +build/ +.dist/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..169cb26 --- /dev/null +++ b/LICENSE @@ -0,0 +1,13 @@ +Copyright 2026 NetBox Export contributors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/README.md b/README.md new file mode 100644 index 0000000..fdd82bb --- /dev/null +++ b/README.md @@ -0,0 +1,92 @@ +# NetBox-Export + +NetBox-Export ist ein Plugin für NetBox 4.6.x. Es exportiert einen abgegrenzten +Mandanten- oder Standortbereich als portables ZIP-Archiv und importiert ihn in +eine zweite NetBox-Instanz. + +Unterstützte Startpunkte: + +- Mandantengruppe einschließlich Untergruppen und Mandanten +- einzelner Mandant +- Region einschließlich Unterregionen und Standorten +- einzelner Standort +- Lokation einschließlich Unterlokationen + +Der Export folgt den Besitzbeziehungen zu DCIM-, IPAM-, Circuit-, +Virtualisierungs-, VPN-, Wireless-, Kontakt-, Tag- und Bilddaten. Benötigte +Stammdaten werden als Abhängigkeiten mitgenommen. Primärschlüssel der +Quellinstanz werden nie direkt als Zielschlüssel verwendet. + +## Installation + +Das Plugin muss auf beiden NetBox-Instanzen installiert sein. + +```bash +cd /opt/netbox +source venv/bin/activate +pip install /path/to/NetBox-Export +``` + +In `configuration.py` ergänzen: + +```python +PLUGINS = [ + "netbox_export", +] + +PLUGINS_CONFIG = { + "netbox_export": { + "max_objects": 50000, + "max_archive_size_mb": 250, + # Auf beiden Instanzen identisch setzen, um Archive zu signieren. + "archive_signing_key": "eine-lange-zufaellige-geheime-zeichenfolge", + }, +} +``` + +Anschließend: + +```bash +cd /opt/netbox/netbox +python manage.py migrate netbox_export +python manage.py collectstatic --no-input +sudo systemctl restart netbox netbox-rq +``` + +Bei einer Docker-Installation wird das Paket in das eigene NetBox-Image +aufgenommen; danach wird der Container mit dem aktivierten Plugin neu gebaut und +die Migration ausgeführt. + +## Verwendung + +Die Oberfläche liegt unter **Plugins > NetBox-Export > Export / Import** und ist +aus Sicherheitsgründen nur für Superuser sichtbar. + +1. Auf Instanz A den Typ und das konkrete Objekt wählen und das ZIP exportieren. +2. Auf Instanz B das ZIP zunächst mit **Nur prüfen** verarbeiten. +3. Nach erfolgreichem Prüflauf **Nur prüfen** deaktivieren, den schreibenden + Import bestätigen und das Archiv erneut hochladen. + +Der Import läuft atomar. Bei einem Fehler werden alle Datenbankänderungen +zurückgerollt. Die Konfliktstrategie **Aktualisieren** nutzt zuerst die dauerhaft +gespeicherte Zuordnung aus Quellinstanz, Modell und Quell-ID; bei einem ersten +Import werden vorhandene Objekte über ihre eindeutigen Fachschlüssel erkannt. + +## Verhalten und Grenzen + +- Quelle und Ziel müssen NetBox 4.6.x und dieselben Plugins/Modelle verwenden. +- Benutzerkonten und Berechtigungen werden nicht exportiert. Referenzen auf + Benutzer oder Gruppen müssen auf dem Ziel bereits eindeutig vorhanden sein. +- Der Import erstellt und aktualisiert Objekte. Zielobjekte, die im Archiv nicht + vorkommen, werden bewusst nicht gelöscht. +- Fehlende Bilddateien werden im Archiv vermerkt, können aber nicht rekonstruiert + werden. +- Große Exporte werden synchron verarbeitet. `max_objects` begrenzt Laufzeit und + Speicherverbrauch. + +## Tests + +```bash +python -m pytest +``` + diff --git a/netbox_export/__init__.py b/netbox_export/__init__.py new file mode 100644 index 0000000..9ffe67f --- /dev/null +++ b/netbox_export/__init__.py @@ -0,0 +1,23 @@ +from typing import ClassVar + +from netbox.plugins import PluginConfig + + +class NetBoxExportConfig(PluginConfig): + name = "netbox_export" + verbose_name = "NetBox-Export" + description = "Portable ZIP export and import for tenants and locations" + version = "0.1.0" + author = "NetBox Export contributors" + base_url = "netbox-export" + min_version = "4.6.0" + max_version = "4.6.99" + required_settings: ClassVar[list[str]] = [] + default_settings: ClassVar[dict] = { + "max_objects": 50000, + "max_archive_size_mb": 250, + "archive_signing_key": "", + } + + +config = NetBoxExportConfig diff --git a/netbox_export/forms.py b/netbox_export/forms.py new file mode 100644 index 0000000..a25dd9f --- /dev/null +++ b/netbox_export/forms.py @@ -0,0 +1,69 @@ +from dcim.models import Location, Region, Site +from django import forms +from tenancy.models import Tenant, TenantGroup + + +class ExportForm(forms.Form): + scope_type = forms.ChoiceField( + label="Exportbereich", + choices=( + ("tenant_group", "Mandantengruppe"), + ("tenant", "Mandant"), + ("region", "Region"), + ("site", "Standort"), + ("location", "Lokation"), + ), + ) + tenant_group = forms.ModelChoiceField( + label="Mandantengruppe", queryset=TenantGroup.objects.all(), required=False + ) + tenant = forms.ModelChoiceField(label="Mandant", queryset=Tenant.objects.all(), required=False) + region = forms.ModelChoiceField(label="Region", queryset=Region.objects.all(), required=False) + site = forms.ModelChoiceField(label="Standort", queryset=Site.objects.all(), required=False) + location = forms.ModelChoiceField(label="Lokation", queryset=Location.objects.all(), required=False) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["tenant"].label_from_instance = lambda obj: f"{obj.group or '-'} / {obj}" + self.fields["location"].label_from_instance = lambda obj: f"{obj.site} / {obj}" + + def clean(self): + cleaned = super().clean() + scope_type = cleaned.get("scope_type") + selected = cleaned.get(scope_type) if scope_type else None + if scope_type and selected is None: + self.add_error(scope_type, "Bitte ein Objekt für den Exportbereich auswählen.") + cleaned["scope_object"] = selected + return cleaned + + +class ImportForm(forms.Form): + archive = forms.FileField( + label="NetBox-Export-Archiv", + widget=forms.ClearableFileInput(attrs={"accept": ".zip,application/zip"}), + ) + conflict_strategy = forms.ChoiceField( + label="Vorhandene Objekte", + choices=( + ("update", "Aktualisieren"), + ("skip", "Überspringen"), + ("fail", "Import abbrechen"), + ), + initial="update", + ) + dry_run = forms.BooleanField( + label="Nur prüfen (keine Änderungen)", + required=False, + initial=True, + help_text="Führt den vollständigen Import aus und setzt die Datenbanktransaktion anschließend zurück.", + ) + confirm_apply = forms.BooleanField( + label="Ich bestätige den schreibenden Import", + required=False, + ) + + def clean(self): + cleaned = super().clean() + if not cleaned.get("dry_run") and not cleaned.get("confirm_apply"): + self.add_error("confirm_apply", "Für einen schreibenden Import ist die Bestätigung erforderlich.") + return cleaned diff --git a/netbox_export/migrations/0001_initial.py b/netbox_export/migrations/0001_initial.py new file mode 100644 index 0000000..6c5d5b6 --- /dev/null +++ b/netbox_export/migrations/0001_initial.py @@ -0,0 +1,47 @@ +import uuid + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + initial = True + + dependencies = [ + ("contenttypes", "0002_remove_content_type_name"), + ] + + operations = [ + migrations.CreateModel( + name="InstanceIdentity", + fields=[ + ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ("singleton", models.BooleanField(default=True, editable=False, unique=True)), + ], + ), + migrations.CreateModel( + name="ImportedObjectMapping", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("source_instance", models.UUIDField()), + ("source_model", models.CharField(max_length=100)), + ("source_object_id", models.CharField(max_length=255)), + ("target_id", models.CharField(max_length=255)), + ("last_imported", models.DateTimeField(auto_now=True)), + ( + "target_type", + models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="contenttypes.contenttype"), + ), + ], + options={ + "indexes": [models.Index(fields=["target_type", "target_id"], name="netbox_exp_target__d4d805_idx")], + "constraints": [ + models.UniqueConstraint( + fields=("source_instance", "source_model", "source_object_id"), + name="netbox_export_unique_source_object", + ) + ], + }, + ), + ] + diff --git a/netbox_export/migrations/__init__.py b/netbox_export/migrations/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/netbox_export/migrations/__init__.py @@ -0,0 +1 @@ + diff --git a/netbox_export/models.py b/netbox_export/models.py new file mode 100644 index 0000000..e5aa0d3 --- /dev/null +++ b/netbox_export/models.py @@ -0,0 +1,39 @@ +import uuid + +from django.contrib.contenttypes.fields import GenericForeignKey +from django.contrib.contenttypes.models import ContentType +from django.db import models + + +class InstanceIdentity(models.Model): + """Stable identity used to correlate repeat exports from this NetBox.""" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + singleton = models.BooleanField(default=True, unique=True, editable=False) + + @classmethod + def local_id(cls): + return cls.objects.get_or_create(singleton=True)[0].pk + + +class ImportedObjectMapping(models.Model): + """Maps an object from another NetBox instance to its local counterpart.""" + + source_instance = models.UUIDField() + source_model = models.CharField(max_length=100) + source_object_id = models.CharField(max_length=255) + target_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) + target_id = models.CharField(max_length=255) + target = GenericForeignKey("target_type", "target_id") + last_imported = models.DateTimeField(auto_now=True) + + class Meta: + constraints = ( + models.UniqueConstraint( + fields=("source_instance", "source_model", "source_object_id"), + name="netbox_export_unique_source_object", + ), + ) + indexes = ( + models.Index(fields=("target_type", "target_id"), name="netbox_exp_target__d4d805_idx"), + ) diff --git a/netbox_export/navigation.py b/netbox_export/navigation.py new file mode 100644 index 0000000..a2c8991 --- /dev/null +++ b/netbox_export/navigation.py @@ -0,0 +1,10 @@ +from netbox.plugins import PluginMenuItem + +menu_items = ( + PluginMenuItem( + link="plugins:netbox_export:dashboard", + link_text="Export / Import", + staff_only=True, + ), +) + diff --git a/netbox_export/services/__init__.py b/netbox_export/services/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/netbox_export/services/__init__.py @@ -0,0 +1 @@ + diff --git a/netbox_export/services/archive.py b/netbox_export/services/archive.py new file mode 100644 index 0000000..98a9668 --- /dev/null +++ b/netbox_export/services/archive.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import hashlib +import hmac +import io +import json +import posixpath +import zipfile +from collections.abc import Iterable +from dataclasses import dataclass +from typing import BinaryIO + +from django.core.serializers.json import DjangoJSONEncoder + +from .exceptions import ArchiveValidationError + +FORMAT_NAME = "netbox-export" +FORMAT_VERSION = 1 +MANIFEST_NAME = "manifest.json" +OBJECTS_NAME = "objects.ndjson" + + +def _json_bytes(value) -> bytes: + return json.dumps( + value, + cls=DjangoJSONEncoder, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def _signature(manifest: dict, signing_key: str) -> str: + signed_manifest = {key: value for key, value in manifest.items() if key != "signature"} + return hmac.new(signing_key.encode("utf-8"), _json_bytes(signed_manifest), hashlib.sha256).hexdigest() + + +def build_archive(manifest: dict, records: Iterable[dict], assets: dict[str, bytes], signing_key: str = "") -> bytes: + normalized_assets = {} + for path, content in assets.items(): + safe_path = _safe_member_name(path) + if safe_path in normalized_assets: + raise ArchiveValidationError(f"Doppelter Dateipfad im Archiv: {safe_path}") + normalized_assets[safe_path] = content + assets = normalized_assets + object_data = b"\n".join(_json_bytes(record) for record in records) + b"\n" + checksum = hashlib.sha256(object_data).hexdigest() + manifest = { + **manifest, + "format": FORMAT_NAME, + "format_version": FORMAT_VERSION, + "objects_sha256": checksum, + "assets_sha256": { + path: hashlib.sha256(content).hexdigest() + for path, content in sorted(assets.items()) + }, + } + if signing_key: + manifest["signature"] = _signature(manifest, signing_key) + + output = io.BytesIO() + with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive: + archive.writestr(MANIFEST_NAME, _json_bytes(manifest)) + archive.writestr(OBJECTS_NAME, object_data) + for path, content in sorted(assets.items()): + archive.writestr(_safe_member_name(path), content) + return output.getvalue() + + +def _safe_member_name(name: str) -> str: + normalized = posixpath.normpath(name.replace("\\", "/")) + if normalized.startswith(("../", "/")) or normalized in ("", ".", ".."): + raise ArchiveValidationError(f"Unsicherer Dateipfad im Archiv: {name}") + return normalized + + +@dataclass(frozen=True) +class ParsedArchive: + manifest: dict + records: list[dict] + assets: dict[str, bytes] + warnings: list[str] + + +def read_archive( + source: bytes | BinaryIO, + *, + max_size: int, + max_objects: int, + signing_key: str = "", +) -> ParsedArchive: + if isinstance(source, bytes): + raw = source + else: + raw = source.read(max_size + 1) + if len(raw) > max_size: + raise ArchiveValidationError("Das Archiv überschreitet die konfigurierte Maximalgröße.") + + try: + archive = zipfile.ZipFile(io.BytesIO(raw)) + except zipfile.BadZipFile as exc: + raise ArchiveValidationError("Die hochgeladene Datei ist kein gültiges ZIP-Archiv.") from exc + + with archive: + names = {} + for info in archive.infolist(): + name = _safe_member_name(info.filename) + if name in names: + raise ArchiveValidationError(f"Doppelter Dateipfad im Archiv: {name}") + names[name] = info + if len(names) > max_objects * 5 + 2: + raise ArchiveValidationError("Das Archiv enthält zu viele Dateien.") + if MANIFEST_NAME not in names or OBJECTS_NAME not in names: + raise ArchiveValidationError("manifest.json oder objects.ndjson fehlt im Archiv.") + total_size = sum(info.file_size for info in names.values()) + if total_size > max_size * 4: + raise ArchiveValidationError("Der entpackte Archivinhalt ist zu groß.") + if any(info.file_size > 0 and info.compress_size * 200 < info.file_size for info in names.values()): + raise ArchiveValidationError("Das Archiv weist ein unzulässiges Kompressionsverhältnis auf.") + + try: + manifest = json.loads(archive.read(MANIFEST_NAME)) + object_data = archive.read(OBJECTS_NAME) + except (KeyError, ValueError, UnicodeDecodeError) as exc: + raise ArchiveValidationError("Die Metadaten im Archiv sind ungültig.") from exc + + if manifest.get("format") != FORMAT_NAME or manifest.get("format_version") != FORMAT_VERSION: + raise ArchiveValidationError("Das Archivformat oder dessen Version wird nicht unterstützt.") + checksum = hashlib.sha256(object_data).hexdigest() + if not hmac.compare_digest(checksum, str(manifest.get("objects_sha256", ""))): + raise ArchiveValidationError("Die Objektdatei stimmt nicht mit ihrer Prüfsumme überein.") + + warnings = [] + signature = manifest.get("signature") + if signing_key: + expected = _signature(manifest, signing_key) + if not signature or not hmac.compare_digest(expected, signature): + raise ArchiveValidationError("Die Archivsignatur ist ungültig oder fehlt.") + elif signature: + warnings.append("Das Archiv ist signiert, aber ohne konfigurierten Schlüssel nicht verifiziert.") + else: + warnings.append("Das Archiv ist nicht signiert.") + + records = [] + try: + for line in object_data.splitlines(): + if line: + records.append(json.loads(line)) + if len(records) > max_objects: + raise ArchiveValidationError("Das Archiv enthält zu viele Objekte.") + except (ValueError, UnicodeDecodeError) as exc: + raise ArchiveValidationError("objects.ndjson enthält ungültiges JSON.") from exc + + assets = { + name: archive.read(name) + for name in names + if name not in (MANIFEST_NAME, OBJECTS_NAME) + } + declared_assets = manifest.get("assets_sha256", {}) + if set(declared_assets) != set(assets): + raise ArchiveValidationError("Die Dateiliste stimmt nicht mit dem Manifest überein.") + for name, content in assets.items(): + actual = hashlib.sha256(content).hexdigest() + if not hmac.compare_digest(actual, str(declared_assets.get(name, ""))): + raise ArchiveValidationError(f"Die Prüfsumme der Datei {name} ist ungültig.") + if manifest.get("object_count") != len(records): + raise ArchiveValidationError("Die Objektanzahl stimmt nicht mit dem Manifest überein.") + return ParsedArchive(manifest=manifest, records=records, assets=assets, warnings=warnings) diff --git a/netbox_export/services/codec.py b/netbox_export/services/codec.py new file mode 100644 index 0000000..79e2eba --- /dev/null +++ b/netbox_export/services/codec.py @@ -0,0 +1,230 @@ +from __future__ import annotations + +import base64 +import datetime +import decimal +import uuid +from pathlib import PurePosixPath + +from django.contrib.contenttypes.fields import GenericForeignKey +from django.db import models + +SKIP_FIELD_NAMES = { + "id", + "created", + "last_updated", + "lft", + "rght", + "tree_id", + "level", +} + + +def object_key(obj) -> str: + return f"{obj._meta.label_lower}:{obj.pk}" + + +def model_label(model) -> str: + return model._meta.label_lower + + +def encode_scalar(value): + if value is None or isinstance(value, (bool, int, float, str)): + return value + if isinstance(value, decimal.Decimal): + return {"$type": "decimal", "value": str(value)} + if isinstance(value, uuid.UUID): + return {"$type": "uuid", "value": str(value)} + if isinstance(value, datetime.datetime): + return {"$type": "datetime", "value": value.isoformat()} + if isinstance(value, datetime.date): + return {"$type": "date", "value": value.isoformat()} + if isinstance(value, datetime.time): + return {"$type": "time", "value": value.isoformat()} + if isinstance(value, datetime.timedelta): + return {"$type": "duration", "value": value.total_seconds()} + if isinstance(value, bytes): + return {"$type": "bytes", "value": base64.b64encode(value).decode("ascii")} + if isinstance(value, (list, tuple)): + return [encode_scalar(item) for item in value] + if isinstance(value, dict): + return {str(key): encode_scalar(item) for key, item in value.items()} + return str(value) + + +def decode_scalar(value): + if isinstance(value, list): + return [decode_scalar(item) for item in value] + if not isinstance(value, dict) or "$type" not in value: + if isinstance(value, dict): + return {key: decode_scalar(item) for key, item in value.items()} + return value + kind = value["$type"] + raw = value.get("value") + decoders = { + "decimal": decimal.Decimal, + "uuid": uuid.UUID, + "datetime": datetime.datetime.fromisoformat, + "date": datetime.date.fromisoformat, + "time": datetime.time.fromisoformat, + "duration": lambda item: datetime.timedelta(seconds=item), + "bytes": lambda item: base64.b64decode(item.encode("ascii")), + } + return decoders[kind](raw) + + +def generic_foreign_keys(model) -> list[GenericForeignKey]: + return [field for field in model._meta.private_fields if isinstance(field, GenericForeignKey)] + + +def external_identity(obj) -> dict: + label = obj._meta.label_lower + if label in ("contenttypes.contenttype", "core.objecttype"): + return {"model": label, "lookup": {"app_label": obj.app_label, "model": obj.model}} + for field_name in ("username", "slug", "name"): + if hasattr(obj, field_name): + return {"model": label, "lookup": {field_name: encode_scalar(getattr(obj, field_name))}} + return {"model": label, "lookup": {"pk": encode_scalar(obj.pk)}} + + +def _reference_spec(obj, exported_keys: set[str]): + if object_key(obj) in exported_keys: + return {"ref": object_key(obj)} + return {"external": external_identity(obj)} + + +def _encode_custom_field_data(obj, value: dict, exported_keys: set[str]): + from extras.models import CustomField + + definitions = {field.name: field for field in CustomField.objects.get_for_model(type(obj))} + encoded = {} + for name, raw_value in value.items(): + custom_field = definitions.get(name) + if ( + custom_field is None + or custom_field.type not in ("object", "multiobject") + or not custom_field.related_object_type + or raw_value in (None, "", []) + ): + encoded[name] = encode_scalar(raw_value) + continue + target_model = custom_field.related_object_type.model_class() + if custom_field.type == "object": + try: + target = target_model._default_manager.get(pk=raw_value) + except target_model.DoesNotExist: + encoded[name] = None + else: + encoded[name] = {"$type": "object_ref", "value": _reference_spec(target, exported_keys)} + else: + targets = {str(item.pk): item for item in target_model._default_manager.filter(pk__in=raw_value)} + encoded[name] = { + "$type": "multiobject_ref", + "value": [ + _reference_spec(targets[str(pk)], exported_keys) + for pk in raw_value + if str(pk) in targets + ], + } + return encoded + + +def _encode_custom_field_default(custom_field, value, exported_keys: set[str]): + if ( + custom_field.type not in ("object", "multiobject") + or not custom_field.related_object_type + or value in (None, "", []) + ): + return encode_scalar(value) + target_model = custom_field.related_object_type.model_class() + if custom_field.type == "object": + try: + target = target_model._default_manager.get(pk=value) + except target_model.DoesNotExist: + return None + return {"$type": "object_ref", "value": _reference_spec(target, exported_keys)} + targets = {str(item.pk): item for item in target_model._default_manager.filter(pk__in=value)} + return { + "$type": "multiobject_ref", + "value": [ + _reference_spec(targets[str(pk)], exported_keys) + for pk in value + if str(pk) in targets + ], + } + + +def serialize_object(obj, exported_keys: set[str], assets: dict[str, bytes]) -> dict: + record_id = object_key(obj) + gfk_fields = generic_foreign_keys(type(obj)) + gfk_storage = {name for field in gfk_fields for name in (field.ct_field, field.fk_field)} + scalars = {} + relations = {} + files = {} + + for field in obj._meta.concrete_fields: + if field.primary_key or field.name.startswith("_") or field.name in SKIP_FIELD_NAMES or field.name in gfk_storage: + continue + if isinstance(field, (models.ForeignKey, models.OneToOneField)): + related = getattr(obj, field.name, None) + if related is None: + relations[field.name] = None + elif object_key(related) in exported_keys: + relations[field.name] = {"ref": object_key(related)} + else: + relations[field.name] = {"external": external_identity(related)} + continue + if isinstance(field, models.FileField): + file_value = getattr(obj, field.name) + if not file_value: + files[field.name] = None + continue + filename = PurePosixPath(str(file_value.name)).name + archive_path = f"assets/{record_id.replace(':', '_')}/{field.name}/{filename}" + try: + with file_value.open("rb") as source: + assets[archive_path] = source.read() + except (FileNotFoundError, OSError, ValueError): + files[field.name] = {"missing": str(file_value.name)} + else: + files[field.name] = {"path": archive_path, "name": str(file_value.name)} + continue + value = field.value_from_object(obj) + if field.name == "custom_field_data" and isinstance(value, dict): + scalars[field.name] = _encode_custom_field_data(obj, value, exported_keys) + elif obj._meta.label_lower == "extras.customfield" and field.name == "default": + scalars[field.name] = _encode_custom_field_default(obj, value, exported_keys) + else: + scalars[field.name] = encode_scalar(value) + + generic_relations = {} + for field in gfk_fields: + related = getattr(obj, field.name, None) + if related is None: + generic_relations[field.name] = None + elif object_key(related) in exported_keys: + generic_relations[field.name] = {"ref": object_key(related)} + else: + generic_relations[field.name] = {"external": external_identity(related)} + + many_to_many = {} + for field in obj._meta.many_to_many: + try: + values = list(getattr(obj, field.name).all()) + except (AttributeError, TypeError): + continue + many_to_many[field.name] = [ + {"ref": object_key(item)} if object_key(item) in exported_keys else {"external": external_identity(item)} + for item in values + ] + + return { + "id": record_id, + "model": obj._meta.label_lower, + "source_pk": str(obj.pk), + "fields": scalars, + "relations": relations, + "generic_relations": generic_relations, + "many_to_many": many_to_many, + "files": files, + } diff --git a/netbox_export/services/exceptions.py b/netbox_export/services/exceptions.py new file mode 100644 index 0000000..9c7060a --- /dev/null +++ b/netbox_export/services/exceptions.py @@ -0,0 +1,15 @@ +class ExportImportError(Exception): + """Base class for errors safe to present in the UI.""" + + +class ArchiveValidationError(ExportImportError): + pass + + +class GraphLimitError(ExportImportError): + pass + + +class ImportConflictError(ExportImportError): + pass + diff --git a/netbox_export/services/exporter.py b/netbox_export/services/exporter.py new file mode 100644 index 0000000..3bf61d2 --- /dev/null +++ b/netbox_export/services/exporter.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from collections import Counter +from datetime import UTC, datetime + +from django.conf import settings + +from netbox_export.models import InstanceIdentity + +from .archive import build_archive +from .codec import serialize_object +from .graph import ObjectGraph, seed_scope + + +def export_scope(scope_type: str, scope_id: int, *, max_objects: int, signing_key: str = ""): + root, seeds = seed_scope(scope_type, scope_id) + graph = ObjectGraph(max_objects=max_objects).collect(seeds) + objects = graph.objects + assets = {} + records = [ + serialize_object(obj, set(objects), assets) + for _, obj in sorted(objects.items()) + ] + counts = Counter(record["model"] for record in records) + manifest = { + "created_at": datetime.now(UTC).isoformat(), + "source_instance": str(InstanceIdentity.local_id()), + "source_netbox_version": getattr(getattr(settings, "RELEASE", None), "version", "4.6"), + "plugin_version": "0.1.0", + "scope": { + "type": scope_type, + "source_pk": str(scope_id), + "label": str(root), + }, + "object_count": len(records), + "member_count": len(graph.members), + "dependency_count": len(graph.dependencies), + "models": dict(sorted(counts.items())), + } + return build_archive(manifest, records, assets, signing_key=signing_key), manifest diff --git a/netbox_export/services/graph.py b/netbox_export/services/graph.py new file mode 100644 index 0000000..81ced30 --- /dev/null +++ b/netbox_export/services/graph.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +from collections import deque + +from django.apps import apps +from django.contrib.contenttypes.fields import GenericForeignKey +from django.contrib.contenttypes.models import ContentType +from django.db import models + +from .codec import object_key +from .exceptions import GraphLimitError + +EXCLUDED_APP_LABELS = { + "account", + "admin", + "auth", + "contenttypes", + "sessions", + "users", + "netbox_export", +} +EXCLUDED_MODELS = { + "core.job", + "core.objectchange", + "extras.eventrule", + "extras.journalentry", + "extras.notification", + "extras.notificationgroup", + "extras.savedfilter", + "extras.subscription", +} +SCOPE_LINK_FIELDS = {"tenant", "site", "location", "region"} +PEER_CONTAINER_MODELS = {"dcim.cable", "circuits.circuit", "circuits.virtualcircuit"} + + +def is_exportable_model(model) -> bool: + opts = model._meta + return bool( + opts.managed + and not opts.abstract + and not opts.proxy + and not opts.auto_created + and opts.app_label not in EXCLUDED_APP_LABELS + and opts.label_lower not in EXCLUDED_MODELS + ) + + +def exportable_models(): + return tuple(model for model in apps.get_models() if is_exportable_model(model)) + + +def _descendants(obj): + if hasattr(obj, "get_descendants"): + return list(obj.get_descendants(include_self=True)) + return [obj] + + +def seed_scope(scope_type: str, scope_id: int): + from dcim.models import Location, Region, Site + from tenancy.models import Tenant, TenantGroup + + models_by_scope = { + "tenant_group": TenantGroup, + "tenant": Tenant, + "region": Region, + "site": Site, + "location": Location, + } + model = models_by_scope[scope_type] + root = model.objects.get(pk=scope_id) + seeds = _descendants(root) + + if scope_type == "tenant_group": + group_ids = [obj.pk for obj in seeds] + seeds.extend(Tenant.objects.filter(group_id__in=group_ids)) + elif scope_type == "region": + region_ids = [obj.pk for obj in seeds] + seeds.extend(Site.objects.filter(region_id__in=region_ids)) + return root, seeds + + +class ObjectGraph: + """Collect scoped objects first and their forward dependencies second.""" + + def __init__(self, max_objects: int): + self.max_objects = max_objects + self.members: dict[str, models.Model] = {} + self.dependencies: dict[str, models.Model] = {} + + @property + def objects(self) -> dict[str, models.Model]: + return {**self.members, **self.dependencies} + + def _check_limit(self): + if len(self.objects) > self.max_objects: + raise GraphLimitError( + f"Der Export würde mehr als {self.max_objects} Objekte enthalten. " + "Bitte den Bereich verkleinern oder max_objects erhöhen." + ) + + def add_member(self, obj) -> bool: + key = object_key(obj) + if key in self.members: + return False + self.dependencies.pop(key, None) + self.members[key] = obj + self._check_limit() + return True + + def add_dependency(self, obj) -> bool: + key = object_key(obj) + if key in self.members or key in self.dependencies or not is_exportable_model(type(obj)): + return False + self.dependencies[key] = obj + self._check_limit() + return True + + def collect(self, seeds): + queue = deque() + for obj in seeds: + if self.add_member(obj): + queue.append(obj) + + models_to_scan = exportable_models() + while True: + while queue: + parent = queue.popleft() + parent_model = type(parent) + parent_ct = ContentType.objects.get_for_model(parent_model) + for candidate_model in models_to_scan: + query = models.Q() + for field in candidate_model._meta.concrete_fields: + if not isinstance(field, (models.ForeignKey, models.OneToOneField)): + continue + if field.remote_field.model is not parent_model: + continue + if ( + field.remote_field.on_delete not in (models.CASCADE, models.PROTECT) + and field.name not in SCOPE_LINK_FIELDS + ): + continue + query |= models.Q(**{field.attname: parent.pk}) + for field in candidate_model._meta.private_fields: + if isinstance(field, GenericForeignKey): + query |= models.Q(**{field.ct_field: parent_ct, field.fk_field: parent.pk}) + if not query: + continue + for child in candidate_model.objects.filter(query).distinct().iterator(): + if self.add_member(child): + queue.append(child) + + promoted = False + for obj in list(self.members.values()): + for field in obj._meta.concrete_fields: + if not isinstance(field, (models.ForeignKey, models.OneToOneField)): + continue + related = getattr(obj, field.name, None) + if related is None or related._meta.label_lower not in PEER_CONTAINER_MODELS: + continue + if self.add_member(related): + queue.append(related) + promoted = True + if not promoted: + break + + dependency_queue = deque(self.members.values()) + scanned = set() + while dependency_queue: + obj = dependency_queue.popleft() + key = object_key(obj) + if key in scanned: + continue + scanned.add(key) + related_objects = [] + for field in obj._meta.concrete_fields: + if isinstance(field, (models.ForeignKey, models.OneToOneField)): + related = getattr(obj, field.name, None) + if related is not None: + related_objects.append(related) + for field in obj._meta.private_fields: + if isinstance(field, GenericForeignKey): + related = getattr(obj, field.name, None) + if related is not None: + related_objects.append(related) + for field in obj._meta.many_to_many: + try: + related_objects.extend(getattr(obj, field.name).all()) + except (AttributeError, TypeError): + pass + if hasattr(obj, "custom_field_data"): + from extras.models import CustomField + + custom_fields = list(CustomField.objects.get_for_model(type(obj))) + related_objects.extend(custom_fields) + for custom_field in custom_fields: + if custom_field.type not in ("object", "multiobject") or not custom_field.related_object_type: + continue + raw_value = obj.custom_field_data.get(custom_field.name) + if raw_value in (None, "", []): + continue + target_model = custom_field.related_object_type.model_class() + target_ids = raw_value if custom_field.type == "multiobject" else [raw_value] + related_objects.extend(target_model._default_manager.filter(pk__in=target_ids)) + if ( + obj._meta.label_lower == "extras.customfield" + and obj.type in ("object", "multiobject") + and obj.related_object_type + and obj.default not in (None, "", []) + ): + target_model = obj.related_object_type.model_class() + target_ids = obj.default if obj.type == "multiobject" else [obj.default] + related_objects.extend(target_model._default_manager.filter(pk__in=target_ids)) + for related in related_objects: + if self.add_dependency(related): + dependency_queue.append(related) + return self diff --git a/netbox_export/services/importer.py b/netbox_export/services/importer.py new file mode 100644 index 0000000..9081410 --- /dev/null +++ b/netbox_export/services/importer.py @@ -0,0 +1,451 @@ +from __future__ import annotations + +import logging +import uuid +from dataclasses import dataclass +from dataclasses import field as dataclass_field + +from django.apps import apps +from django.conf import settings +from django.contrib.contenttypes.models import ContentType +from django.core.files.base import ContentFile +from django.db import IntegrityError, models, transaction + +from netbox_export.models import ImportedObjectMapping + +from .archive import ParsedArchive +from .codec import SKIP_FIELD_NAMES, decode_scalar, generic_foreign_keys +from .exceptions import ArchiveValidationError, ExportImportError, ImportConflictError + +EXPLICIT_IDENTITIES = { + "tenancy.tenantgroup": ("slug",), + "tenancy.tenant": ("group", "slug"), + "dcim.region": ("parent", "slug"), + "dcim.sitegroup": ("parent", "slug"), + "dcim.site": ("slug",), + "dcim.location": ("site", "parent", "slug"), + "dcim.rack": ("site", "location", "name"), + "dcim.device": ("site", "tenant", "name"), + "dcim.interface": ("device", "name"), + "dcim.consoleport": ("device", "name"), + "dcim.consoleserverport": ("device", "name"), + "dcim.powerport": ("device", "name"), + "dcim.poweroutlet": ("device", "name"), + "dcim.devicebay": ("device", "name"), + "dcim.modulebay": ("device", "name"), + "dcim.inventoryitem": ("device", "parent", "name"), + "ipam.prefix": ("vrf", "prefix"), + "ipam.ipaddress": ("vrf", "address"), + "ipam.vlan": ("group", "vid"), + "circuits.circuit": ("provider", "cid"), + "virtualization.virtualmachine": ("cluster", "tenant", "name"), + "virtualization.vminterface": ("virtual_machine", "name"), +} + + +class IdentityNotReady(Exception): + pass + + +logger = logging.getLogger(__name__) + + +@dataclass +class ImportReport: + dry_run: bool + created: int = 0 + updated: int = 0 + skipped: int = 0 + mapped: int = 0 + models: dict[str, dict[str, int]] = dataclass_field(default_factory=dict) + warnings: list[str] = dataclass_field(default_factory=list) + + def add(self, model: str, action: str): + setattr(self, action, getattr(self, action) + 1) + counters = self.models.setdefault(model, {"created": 0, "updated": 0, "skipped": 0}) + counters[action] += 1 + + +def _model_for(label: str): + try: + model = apps.get_model(label) + except (LookupError, ValueError) as exc: + raise ArchiveValidationError(f"Das Modell {label} ist auf der Zielinstanz nicht installiert.") from exc + if model is None: + raise ArchiveValidationError(f"Das Modell {label} ist auf der Zielinstanz nicht installiert.") + return model + + +def _external_object(spec: dict): + model = _model_for(spec["model"]) + lookup = {key: decode_scalar(value) for key, value in spec.get("lookup", {}).items()} + try: + return model._default_manager.get(**lookup) + except model.DoesNotExist as exc: + raise ArchiveValidationError( + f"Externe Referenz fehlt: {spec['model']} mit {lookup}." + ) from exc + except model.MultipleObjectsReturned as exc: + raise ArchiveValidationError( + f"Externe Referenz ist nicht eindeutig: {spec['model']} mit {lookup}." + ) from exc + + +def _resolve_reference(spec, resolved: dict[str, models.Model]): + if spec is None: + return None, True + if "ref" in spec: + return resolved.get(spec["ref"]), spec["ref"] in resolved + if "external" in spec: + return _external_object(spec["external"]), True + raise ArchiveValidationError("Eine Objektreferenz im Archiv ist ungültig.") + + +def _decode_archived_value(encoded, resolved): + if isinstance(encoded, dict) and encoded.get("$type") == "object_ref": + target, available = _resolve_reference(encoded.get("value"), resolved) + return (target.pk if available and target is not None else None), available + if isinstance(encoded, dict) and encoded.get("$type") == "multiobject_ref": + values = [] + for spec in encoded.get("value", []): + target, available = _resolve_reference(spec, resolved) + if not available: + return None, False + values.append(target.pk) + return values, True + if isinstance(encoded, dict) and "$type" not in encoded: + value = {} + all_available = True + for key, item in encoded.items(): + decoded, available = _decode_archived_value(item, resolved) + value[key] = decoded + all_available &= available + return value, all_available + return decode_scalar(encoded), True + + +def _identity_candidates(model): + explicit = EXPLICIT_IDENTITIES.get(model._meta.label_lower) + if explicit: + yield explicit + for field in model._meta.concrete_fields: + if field.unique and not field.primary_key: + yield (field.name,) + if model._meta.unique_together: + yield from model._meta.unique_together + for constraint in model._meta.constraints: + if isinstance(constraint, models.UniqueConstraint) and constraint.fields: + yield tuple(constraint.fields) + + +def _identity_lookup(model, record, resolved): + scalar_values = record.get("fields", {}) + relation_values = record.get("relations", {}) + generic_values = record.get("generic_relations", {}) + generic_storage = {} + for generic_field in generic_foreign_keys(model): + if generic_field.name not in generic_values: + continue + spec = generic_values[generic_field.name] + generic_storage[generic_field.ct_field] = ("content_type", spec) + generic_storage[generic_field.fk_field] = ("object_id", spec) + for candidate in _identity_candidates(model): + lookup = {} + usable = True + for name in candidate: + if name in scalar_values: + value = decode_scalar(scalar_values[name]) + elif name in relation_values: + value, available = _resolve_reference(relation_values[name], resolved) + if not available: + raise IdentityNotReady + elif name in generic_storage: + value_type, spec = generic_storage[name] + target, available = _resolve_reference(spec, resolved) + if not available: + raise IdentityNotReady + if target is None: + value = None + elif value_type == "content_type": + value = ContentType.objects.get_for_model(target, for_concrete_model=False) + else: + value = target.pk + else: + usable = False + break + if value is None and len(candidate) == 1: + usable = False + break + lookup[name] = value + if usable: + return lookup + return None + + +def _mapped_object(source_instance: uuid.UUID, record: dict, model): + mapping = ImportedObjectMapping.objects.filter( + source_instance=source_instance, + source_model=record["model"], + source_object_id=record["source_pk"], + ).first() + if not mapping: + return None + try: + return model._default_manager.get(pk=mapping.target_id) + except model.DoesNotExist: + mapping.delete() + return None + + +def _find_existing(source_instance, model, record, resolved): + mapped = _mapped_object(source_instance, record, model) + if mapped is not None: + return mapped + lookup = _identity_lookup(model, record, resolved) + if not lookup: + return None + try: + return model._default_manager.get(**lookup) + except model.DoesNotExist: + return None + except model.MultipleObjectsReturned as exc: + raise ImportConflictError( + f"Mehrere Zielobjekte passen auf {record['model']} mit {lookup}." + ) from exc + + +def _write_mapping(source_instance, record, obj): + content_type = ContentType.objects.get_for_model(obj, for_concrete_model=False) + ImportedObjectMapping.objects.update_or_create( + source_instance=source_instance, + source_model=record["model"], + source_object_id=record["source_pk"], + defaults={"target_type": content_type, "target_id": str(obj.pk)}, + ) + + +def _field_kwargs(model, record, resolved): + valid_fields = {field.name: field for field in model._meta.concrete_fields} + kwargs = {} + unresolved = [] + unresolved_values = [] + for name, encoded in record.get("fields", {}).items(): + field = valid_fields.get(name) + if ( + not field + or field.primary_key + or name.startswith("_") + or name in SKIP_FIELD_NAMES + or isinstance(field, models.FileField) + ): + continue + if isinstance(field, (models.ForeignKey, models.OneToOneField)): + continue + value, available = _decode_archived_value(encoded, resolved) + if available: + kwargs[name] = value + elif name in ("custom_field_data", "default"): + kwargs[name] = value if value is not None else ({} if name == "custom_field_data" else None) + unresolved_values.append((name, encoded)) + else: + return None, [], [] + for name, spec in record.get("relations", {}).items(): + field = valid_fields.get(name) + if not isinstance(field, (models.ForeignKey, models.OneToOneField)): + continue + value, available = _resolve_reference(spec, resolved) + if available: + kwargs[name] = value + elif field.null: + kwargs[name] = None + unresolved.append((name, spec)) + else: + return None, [], [] + return kwargs, unresolved, unresolved_values + + +def _set_generic_relations(obj, record, resolved, *, allow_deferred: bool): + fields = {field.name: field for field in generic_foreign_keys(type(obj))} + unresolved = [] + for name, spec in record.get("generic_relations", {}).items(): + field = fields.get(name) + if not field: + continue + value, available = _resolve_reference(spec, resolved) + if available: + setattr(obj, name, value) + elif allow_deferred: + ct_field = obj._meta.get_field(field.ct_field) + id_field = obj._meta.get_field(field.fk_field) + if ct_field.null and id_field.null: + setattr(obj, field.ct_field, None) + setattr(obj, field.fk_field, None) + unresolved.append((name, spec)) + else: + return None + else: + return None + return unresolved + + +def _set_files(obj, record, assets, saved_files, *, dry_run: bool): + if dry_run: + return + for name, spec in record.get("files", {}).items(): + if not spec or "path" not in spec or spec["path"] not in assets: + continue + filename = spec.get("name", spec["path"]).replace("\\", "/").rsplit("/", 1)[-1] + file_value = getattr(obj, name) + file_value.save(filename, ContentFile(assets[spec["path"]]), save=False) + saved_files.append((file_value.storage, file_value.name)) + + +def _cleanup_files(saved_files): + for storage, name in reversed(saved_files): + try: + storage.delete(name) + except Exception: + logger.warning("Could not remove rolled-back import file %s", name, exc_info=True) + + +def _apply_m2m(obj, record, resolved): + for name, specs in record.get("many_to_many", {}).items(): + try: + manager = getattr(obj, name) + except AttributeError: + continue + values = [] + for spec in specs: + value, available = _resolve_reference(spec, resolved) + if not available: + raise ArchiveValidationError(f"M2M-Referenz für {record['id']} konnte nicht aufgelöst werden.") + values.append(value) + manager.set(values) + + +def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bool) -> ImportReport: + if conflict_strategy not in ("update", "skip", "fail"): + raise ArchiveValidationError("Unbekannte Konfliktstrategie.") + source_version = str(parsed.manifest.get("source_netbox_version", "")) + target_version = str(getattr(getattr(settings, "RELEASE", None), "version", "")) + if source_version and target_version and source_version.split(".")[:2] != target_version.split(".")[:2]: + raise ArchiveValidationError( + f"NetBox-Versionen sind nicht kompatibel: Quelle {source_version}, Ziel {target_version}." + ) + try: + source_instance = uuid.UUID(parsed.manifest["source_instance"]) + except (KeyError, TypeError, ValueError) as exc: + raise ArchiveValidationError("Die Quellinstanz-Kennung fehlt oder ist ungültig.") from exc + + records = {} + for record in parsed.records: + if not all(key in record for key in ("id", "model", "source_pk")): + raise ArchiveValidationError("Ein Objektdatensatz ist unvollständig.") + if record["id"] in records: + raise ArchiveValidationError(f"Doppelte Objekt-ID im Archiv: {record['id']}") + records[record["id"]] = record + + report = ImportReport(dry_run=dry_run, warnings=list(parsed.warnings)) + resolved = {} + deferred_relations = [] + deferred_generic = [] + deferred_values = [] + writable = set() + saved_files = [] + + try: + with transaction.atomic(): + pending = dict(records) + while pending: + progressed = False + for record_id, record in list(pending.items()): + model = _model_for(record["model"]) + kwargs, unresolved, unresolved_value_fields = _field_kwargs(model, record, resolved) + if kwargs is None: + continue + try: + existing = _find_existing(source_instance, model, record, resolved) + except IdentityNotReady: + continue + if existing is not None and conflict_strategy == "fail": + raise ImportConflictError(f"Zielobjekt existiert bereits: {record_id}") + + if existing is not None and conflict_strategy == "skip": + obj = existing + action = "skipped" + else: + obj = existing or model() + for name, value in kwargs.items(): + setattr(obj, name, value) + generic_unresolved = _set_generic_relations(obj, record, resolved, allow_deferred=True) + if generic_unresolved is None: + continue + _set_files(obj, record, parsed.assets, saved_files, dry_run=dry_run) + obj.save() + action = "updated" if existing is not None else "created" + writable.add(record_id) + deferred_relations.extend((record_id, name, spec) for name, spec in unresolved) + deferred_generic.extend((record_id, name, spec) for name, spec in generic_unresolved) + deferred_values.extend( + (record_id, name, encoded) for name, encoded in unresolved_value_fields + ) + + resolved[record_id] = obj + _write_mapping(source_instance, record, obj) + report.add(record["model"], action) + report.mapped += 1 + pending.pop(record_id) + progressed = True + if not progressed: + blocked = ", ".join(list(pending)[:10]) + raise ArchiveValidationError( + f"Erforderliche Referenzen konnten nicht aufgelöst werden: {blocked}" + ) + + for record_id, name, spec in deferred_relations: + if record_id not in writable: + continue + value, available = _resolve_reference(spec, resolved) + if not available: + raise ArchiveValidationError(f"Referenz {name} für {record_id} konnte nicht aufgelöst werden.") + obj = resolved[record_id] + setattr(obj, name, value) + obj.save(update_fields=[name]) + + for record_id, name, spec in deferred_generic: + if record_id not in writable: + continue + value, available = _resolve_reference(spec, resolved) + if not available: + raise ArchiveValidationError(f"Generische Referenz {name} für {record_id} fehlt.") + obj = resolved[record_id] + setattr(obj, name, value) + field = next(field for field in generic_foreign_keys(type(obj)) if field.name == name) + obj.save(update_fields=[field.ct_field, field.fk_field]) + + for record_id, name, encoded in deferred_values: + if record_id not in writable: + continue + value, available = _decode_archived_value(encoded, resolved) + if not available: + raise ArchiveValidationError(f"Custom-Field-Referenz {name} für {record_id} fehlt.") + obj = resolved[record_id] + setattr(obj, name, value) + obj.save(update_fields=[name]) + + for record_id, record in records.items(): + if record_id in writable: + _apply_m2m(resolved[record_id], record, resolved) + + if dry_run: + transaction.set_rollback(True) + except (IntegrityError, ValueError, TypeError) as exc: + _cleanup_files(saved_files) + raise ArchiveValidationError(f"Der Import wurde zurückgerollt: {exc}") from exc + except ExportImportError: + _cleanup_files(saved_files) + raise + except Exception as exc: + _cleanup_files(saved_files) + raise ArchiveValidationError(f"Der Import wurde zurückgerollt: {exc}") from exc + return report diff --git a/netbox_export/templates/netbox_export/dashboard.html b/netbox_export/templates/netbox_export/dashboard.html new file mode 100644 index 0000000..88682da --- /dev/null +++ b/netbox_export/templates/netbox_export/dashboard.html @@ -0,0 +1,117 @@ +{% extends 'base/layout.html' %} +{% load form_helpers %} + +{% block title %}NetBox-Export{% endblock title %} + +{% block header %} +
+
+
+
+

NetBox-Export

+
+
+
+
+{% endblock header %} + +{% block content %} + {% if report %} + +
+ + + + + + {% for model, counts in report.models.items %} + + + + + + + {% endfor %} + +
ModellErstelltAktualisiertÜbersprungen
{{ model }}{{ counts.created }}{{ counts.updated }}{{ counts.skipped }}
+
+ {% endif %} + +
+
+
+
+

Export

+
+
+ {% csrf_token %} + +
+ {% if export_form.non_field_errors %} +
{{ export_form.non_field_errors }}
+ {% endif %} + {% render_form export_form %} +
+
+ +
+
+
+
+ +
+
+
+

Import

+
+
+ {% csrf_token %} + +
+ {% if import_form.non_field_errors %} +
{{ import_form.non_field_errors }}
+ {% endif %} + {% render_form import_form %} +
+
+ +
+
+
+
+
+{% endblock content %} + +{% block javascript %} + {{ block.super }} + +{% endblock javascript %} diff --git a/netbox_export/urls.py b/netbox_export/urls.py new file mode 100644 index 0000000..e3a4d7e --- /dev/null +++ b/netbox_export/urls.py @@ -0,0 +1,10 @@ +from django.urls import path + +from . import views + +app_name = "netbox_export" + +urlpatterns = [ + path("", views.DashboardView.as_view(), name="dashboard"), +] + diff --git a/netbox_export/views.py b/netbox_export/views.py new file mode 100644 index 0000000..85ee017 --- /dev/null +++ b/netbox_export/views.py @@ -0,0 +1,88 @@ +from django.contrib.auth.mixins import UserPassesTestMixin +from django.http import HttpResponse +from django.shortcuts import render +from django.utils.text import slugify +from django.views import View +from netbox.plugins import get_plugin_config + +from .forms import ExportForm, ImportForm +from .services.archive import read_archive +from .services.exceptions import ExportImportError +from .services.exporter import export_scope +from .services.importer import import_archive + + +class DashboardView(UserPassesTestMixin, View): + template_name = "netbox_export/dashboard.html" + raise_exception = True + + def test_func(self): + return self.request.user.is_authenticated and self.request.user.is_superuser + + def _context(self, export_form=None, import_form=None, report=None, manifest=None): + return { + "export_form": export_form or ExportForm(prefix="export"), + "import_form": import_form or ImportForm(prefix="import"), + "report": report, + "manifest": manifest, + } + + def get(self, request): + return render(request, self.template_name, self._context()) + + def post(self, request): + action = request.POST.get("action") + if action == "export": + return self._export(request) + if action == "import": + return self._import(request) + return render(request, self.template_name, self._context(), status=400) + + def _export(self, request): + form = ExportForm(request.POST, prefix="export") + if not form.is_valid(): + return render(request, self.template_name, self._context(export_form=form), status=400) + scope = form.cleaned_data["scope_object"] + try: + payload, _ = export_scope( + form.cleaned_data["scope_type"], + scope.pk, + max_objects=int(get_plugin_config("netbox_export", "max_objects", 50000)), + signing_key=get_plugin_config("netbox_export", "archive_signing_key", ""), + ) + except ExportImportError as exc: + form.add_error(None, str(exc)) + return render(request, self.template_name, self._context(export_form=form), status=400) + + filename = f"netbox-export-{form.cleaned_data['scope_type']}-{slugify(str(scope)) or scope.pk}.zip" + response = HttpResponse(payload, content_type="application/zip") + response["Content-Disposition"] = f'attachment; filename="{filename}"' + response["X-Content-Type-Options"] = "nosniff" + return response + + def _import(self, request): + form = ImportForm(request.POST, request.FILES, prefix="import") + if not form.is_valid(): + return render(request, self.template_name, self._context(import_form=form), status=400) + max_size = int(get_plugin_config("netbox_export", "max_archive_size_mb", 250)) * 1024 * 1024 + try: + parsed = read_archive( + form.cleaned_data["archive"], + max_size=max_size, + max_objects=int(get_plugin_config("netbox_export", "max_objects", 50000)), + signing_key=get_plugin_config("netbox_export", "archive_signing_key", ""), + ) + report = import_archive( + parsed, + conflict_strategy=form.cleaned_data["conflict_strategy"], + dry_run=form.cleaned_data["dry_run"], + ) + except ExportImportError as exc: + form.add_error(None, str(exc)) + return render(request, self.template_name, self._context(import_form=form), status=400) + return render( + request, + self.template_name, + self._context(import_form=ImportForm(prefix="import"), report=report, manifest=parsed.manifest), + ) + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..bedc9e5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,30 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "netbox-export" +version = "0.1.0" +description = "Portable ZIP export and import for scoped NetBox data" +readme = "README.md" +requires-python = ">=3.12" +license = {text = "Apache-2.0"} +authors = [{name = "NetBox Export contributors"}] +classifiers = [ + "Framework :: Django", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.12", +] + +[tool.setuptools.packages.find] +include = ["netbox_export*"] + +[tool.setuptools.package-data] +netbox_export = ["templates/**/*.html"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" + +[tool.ruff.lint.per-file-ignores] +"netbox_export/migrations/*.py" = ["RUF012"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a9488fe --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,27 @@ +import sys +import types + +from django.conf import settings + +if not settings.configured: + settings.configure( + INSTALLED_APPS=["django.contrib.contenttypes"], + SECRET_KEY="tests", + ) + +netbox_module = types.ModuleType("netbox") +plugins_module = types.ModuleType("netbox.plugins") + + +class PluginConfig: + pass + + +plugins_module.PluginConfig = PluginConfig +netbox_module.plugins = plugins_module +sys.modules.setdefault("netbox", netbox_module) +sys.modules.setdefault("netbox.plugins", plugins_module) + +import django + +django.setup() diff --git a/tests/test_archive.py b/tests/test_archive.py new file mode 100644 index 0000000..d7ef675 --- /dev/null +++ b/tests/test_archive.py @@ -0,0 +1,99 @@ +import io +import json +import zipfile + +import pytest + +from netbox_export.services.archive import build_archive, read_archive +from netbox_export.services.exceptions import ArchiveValidationError + + +def manifest(): + return { + "source_instance": "b89196f8-3d87-466a-9278-f68e22d6d2cc", + "scope": {"type": "site", "source_pk": "1", "label": "Berlin"}, + "object_count": 1, + } + + +def records(): + return [ + { + "id": "dcim.site:1", + "model": "dcim.site", + "source_pk": "1", + "fields": {"name": "Berlin"}, + } + ] + + +def test_archive_round_trip(): + payload = build_archive(manifest(), records(), {"assets/a.txt": b"data"}) + + parsed = read_archive(payload, max_size=1024 * 1024, max_objects=10) + + assert parsed.records == records() + assert parsed.assets == {"assets/a.txt": b"data"} + assert parsed.warnings == ["Das Archiv ist nicht signiert."] + + +def test_signed_archive_requires_matching_key(): + payload = build_archive(manifest(), records(), {}, signing_key="secret-a") + + with pytest.raises(ArchiveValidationError, match="Archivsignatur"): + read_archive(payload, max_size=1024 * 1024, max_objects=10, signing_key="secret-b") + + +def test_signed_archive_without_local_key_is_not_claimed_as_verified(): + payload = build_archive(manifest(), records(), {}, signing_key="secret-a") + + parsed = read_archive(payload, max_size=1024 * 1024, max_objects=10) + + assert parsed.warnings == ["Das Archiv ist signiert, aber ohne konfigurierten Schlüssel nicht verifiziert."] + + +def test_modified_object_stream_is_rejected(): + payload = build_archive(manifest(), records(), {}) + source = zipfile.ZipFile(io.BytesIO(payload)) + output = io.BytesIO() + with source, zipfile.ZipFile(output, "w") as target: + for info in source.infolist(): + content = source.read(info.filename) + if info.filename == "objects.ndjson": + content = content.replace(b"Berlin", b"Hamburg") + target.writestr(info, content) + + with pytest.raises(ArchiveValidationError, match="Prüfsumme"): + read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10) + + +def test_modified_asset_is_rejected(): + payload = build_archive(manifest(), records(), {"assets/a.txt": b"original"}) + source = zipfile.ZipFile(io.BytesIO(payload)) + output = io.BytesIO() + with source, zipfile.ZipFile(output, "w") as target: + for info in source.infolist(): + content = source.read(info.filename) + if info.filename == "assets/a.txt": + content = b"modified" + target.writestr(info, content) + + with pytest.raises(ArchiveValidationError, match="Prüfsumme der Datei"): + read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10) + + + +def test_path_traversal_is_rejected(): + output = io.BytesIO() + with zipfile.ZipFile(output, "w") as archive: + archive.writestr("../manifest.json", json.dumps(manifest())) + + with pytest.raises(ArchiveValidationError, match="Dateipfad"): + read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10) + + +def test_object_limit_is_enforced(): + payload = build_archive(manifest(), records() * 2, {}) + + with pytest.raises(ArchiveValidationError, match="zu viele Objekte"): + read_archive(payload, max_size=1024 * 1024, max_objects=1) diff --git a/tests/test_codec.py b/tests/test_codec.py new file mode 100644 index 0000000..8534da1 --- /dev/null +++ b/tests/test_codec.py @@ -0,0 +1,19 @@ +import datetime +import decimal +import uuid + +from netbox_export.services.codec import decode_scalar, encode_scalar + + +def test_scalar_round_trip(): + value = { + "decimal": decimal.Decimal("12.340"), + "uuid": uuid.UUID("00112233-4455-6677-8899-aabbccddeeff"), + "date": datetime.date(2026, 8, 5), + "datetime": datetime.datetime(2026, 8, 5, 10, 30, tzinfo=datetime.UTC), + "duration": datetime.timedelta(seconds=42), + "bytes": b"binary", + } + + assert decode_scalar(encode_scalar(value)) == value +