274 lines
12 KiB
Python
274 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.http import Http404
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.utils import timezone
|
|
from django.utils.decorators import method_decorator
|
|
from django.views import View
|
|
from django.views.decorators.csrf import csrf_protect
|
|
|
|
from .agent import AgentClient
|
|
from .access import has_store_access
|
|
from .client import StoreClient, StoreClientError
|
|
from .forms import LifecycleConfirmForm
|
|
from .jobs import PluginLifecycleJob
|
|
from .lifecycle import ACTIONS, LifecycleRequest, LifecycleService, installed_distribution_version
|
|
from .models import LifecycleAudit, ManagedPlugin
|
|
from .redaction import redact_data, redact_text
|
|
from .repository import DjangoLifecycleRepository
|
|
from .runtime import RuntimeSettings
|
|
|
|
|
|
class StorePermissionMixin(LoginRequiredMixin):
|
|
permission_name = "netbox_plugin_store.manage_plugin"
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if not has_store_access(request.user, self.permission_name):
|
|
raise PermissionDenied
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
|
|
def _components() -> tuple[RuntimeSettings, StoreClient, DjangoLifecycleRepository]:
|
|
runtime = RuntimeSettings.from_django()
|
|
client = StoreClient(
|
|
runtime.store_url,
|
|
runtime.allowed_store_urls,
|
|
runtime.allowed_artifact_urls,
|
|
api_token=runtime.api_token,
|
|
timeout=runtime.request_timeout,
|
|
)
|
|
return runtime, client, DjangoLifecycleRepository()
|
|
|
|
|
|
def _plugin_state(plugin, active_plugins: set[str]) -> dict:
|
|
installed_version = installed_distribution_version(plugin.package_name)
|
|
enabled = plugin.import_name in active_plugins
|
|
return {
|
|
"installed_version": installed_version,
|
|
"enabled": enabled,
|
|
"installed": bool(installed_version),
|
|
}
|
|
|
|
|
|
class CatalogView(StorePermissionMixin, View):
|
|
def get(self, request):
|
|
from django.conf import settings as django_settings
|
|
|
|
try:
|
|
runtime, client, _ = _components()
|
|
plugins = client.list_plugins()
|
|
active_plugins = set(django_settings.PLUGINS)
|
|
cards = [{"plugin": plugin, **_plugin_state(plugin, active_plugins)} for plugin in plugins]
|
|
error = ""
|
|
except Exception as exc:
|
|
runtime, cards = None, []
|
|
error = redact_text(exc)
|
|
return render(
|
|
request,
|
|
"netbox_plugin_store/catalog.html",
|
|
{"cards": cards, "store_error": error, "runtime": runtime},
|
|
)
|
|
|
|
|
|
class PluginDetailView(StorePermissionMixin, View):
|
|
def get(self, request, slug: str):
|
|
from django.conf import settings as django_settings
|
|
|
|
try:
|
|
runtime, client, _ = _components()
|
|
plugin = client.get_plugin(slug)
|
|
except StoreClientError as exc:
|
|
raise Http404(redact_text(exc)) from exc
|
|
state = _plugin_state(plugin, set(django_settings.PLUGINS))
|
|
installable_releases = plugin.installable_releases(
|
|
runtime.netbox_version,
|
|
require_approval_marker=runtime.execution_mode == "agent",
|
|
)
|
|
actions: list[str] = []
|
|
if not state["installed"]:
|
|
if plugin.approved and installable_releases:
|
|
actions.append("install")
|
|
else:
|
|
if plugin.approved and installable_releases:
|
|
actions.append("update")
|
|
if state["enabled"]:
|
|
actions.append("disable")
|
|
else:
|
|
actions.extend(("enable", "uninstall"))
|
|
releases = [
|
|
{"release": release, "compatible": release.supports(runtime.netbox_version, plugin)}
|
|
for release in plugin.releases
|
|
]
|
|
local_status = ManagedPlugin.objects.filter(slug=slug).first()
|
|
return render(
|
|
request,
|
|
"netbox_plugin_store/detail.html",
|
|
{
|
|
"plugin": plugin,
|
|
"state": state,
|
|
"actions": actions,
|
|
"releases": releases,
|
|
"has_installable_release": bool(installable_releases),
|
|
"local_status": local_status,
|
|
"runtime": runtime,
|
|
},
|
|
)
|
|
|
|
|
|
class LifecycleConfirmView(StorePermissionMixin, View):
|
|
def get(self, request, slug: str, action: str):
|
|
if action not in ACTIONS:
|
|
raise Http404
|
|
runtime, client, _ = _components()
|
|
plugin = client.get_plugin(slug)
|
|
form = LifecycleConfirmForm(plugin=plugin, action=action, runtime=runtime)
|
|
return render(
|
|
request,
|
|
"netbox_plugin_store/confirm.html",
|
|
{"plugin": plugin, "action": action, "form": form, "runtime": runtime},
|
|
)
|
|
|
|
|
|
@method_decorator(csrf_protect, name="dispatch")
|
|
class LifecycleActionView(StorePermissionMixin, View):
|
|
http_method_names = ["post"]
|
|
|
|
def post(self, request, slug: str, action: str):
|
|
if action not in ACTIONS:
|
|
raise Http404
|
|
runtime, client, repository = _components()
|
|
plugin = client.get_plugin(slug)
|
|
form = LifecycleConfirmForm(
|
|
request.POST,
|
|
plugin=plugin,
|
|
action=action,
|
|
runtime=runtime,
|
|
)
|
|
if not form.is_valid():
|
|
return render(
|
|
request,
|
|
"netbox_plugin_store/confirm.html",
|
|
{"plugin": plugin, "action": action, "form": form, "runtime": runtime},
|
|
status=400,
|
|
)
|
|
operation = LifecycleRequest(
|
|
slug=slug,
|
|
action=action,
|
|
version=form.cleaned_data.get("version", ""),
|
|
dry_run=form.cleaned_data["dry_run"],
|
|
)
|
|
if not operation.dry_run and not request.user.is_superuser:
|
|
raise PermissionDenied("Reale Plugin-Lifecycle-Aktionen erfordern einen Superuser.")
|
|
audit_id = repository.create_audit(operation, request.user.pk)
|
|
if operation.dry_run and runtime.background_jobs:
|
|
try:
|
|
PluginLifecycleJob.enqueue(
|
|
user=request.user,
|
|
queue_name=runtime.job_queue,
|
|
audit_id=audit_id,
|
|
slug=slug,
|
|
action=action,
|
|
version=operation.version,
|
|
dry_run=True,
|
|
actor_id=request.user.pk,
|
|
)
|
|
messages.success(request, "Dry-Run wurde in die NetBox-Jobqueue gestellt.")
|
|
return redirect("plugins:netbox_plugin_store:audit-detail", pk=audit_id)
|
|
except Exception as exc:
|
|
if not runtime.synchronous_fallback:
|
|
repository.mark_failed(audit_id, redact_text(exc), {"queue_failed": True})
|
|
messages.error(request, "Dry-Run konnte nicht eingeplant werden.")
|
|
return redirect("plugins:netbox_plugin_store:audit-detail", pk=audit_id)
|
|
service = LifecycleService(runtime, client, repository)
|
|
try:
|
|
result = service.execute(operation, actor_id=request.user.pk, audit_id=audit_id)
|
|
except Exception as exc:
|
|
messages.error(request, f"Lifecycle-Aktion fehlgeschlagen: {redact_text(exc, limit=1_000)}")
|
|
else:
|
|
label = "Dry-Run abgeschlossen" if result.dry_run else "Lifecycle-Aktion abgeschlossen"
|
|
messages.success(request, label + ".")
|
|
return redirect("plugins:netbox_plugin_store:audit-detail", pk=audit_id)
|
|
|
|
|
|
class InstalledStatusView(StorePermissionMixin, View):
|
|
def get(self, request):
|
|
statuses = ManagedPlugin.objects.all()
|
|
return render(request, "netbox_plugin_store/status.html", {"statuses": statuses})
|
|
|
|
|
|
class AuditListView(StorePermissionMixin, View):
|
|
def get(self, request):
|
|
audits = LifecycleAudit.objects.select_related("actor")[:200]
|
|
return render(request, "netbox_plugin_store/audit_list.html", {"audits": audits})
|
|
|
|
|
|
class AuditDetailView(StorePermissionMixin, View):
|
|
def get(self, request, pk: int):
|
|
audit = get_object_or_404(LifecycleAudit.objects.select_related("actor"), pk=pk)
|
|
if audit.status == LifecycleAudit.Status.HANDED_OFF and audit.external_operation_id:
|
|
self._refresh_agent_status(request, audit)
|
|
audit.refresh_from_db()
|
|
return render(request, "netbox_plugin_store/audit_detail.html", {"audit": audit})
|
|
|
|
@staticmethod
|
|
def _refresh_agent_status(request, audit: LifecycleAudit) -> None:
|
|
try:
|
|
runtime = RuntimeSettings.from_django()
|
|
if runtime.execution_mode != "agent" or runtime.agent_socket_path is None:
|
|
return
|
|
body = AgentClient(runtime.agent_socket_path, timeout=runtime.agent_timeout).get_operation(
|
|
audit.external_operation_id
|
|
)
|
|
state = body.get("state") or body.get("status")
|
|
if state in {"succeeded", "completed"}:
|
|
result = body.get("result", body)
|
|
if not isinstance(result, dict):
|
|
result = {"message": "Host agent completed without a structured result."}
|
|
LifecycleAudit.objects.filter(pk=audit.pk).update(
|
|
status=LifecycleAudit.Status.SUCCEEDED,
|
|
result_data=redact_data(result),
|
|
error="",
|
|
completed=timezone.now(),
|
|
)
|
|
status_values = {}
|
|
if "installed_version" in result:
|
|
status_values["installed_version"] = str(result["installed_version"] or "")[:64]
|
|
if "enabled" in result:
|
|
status_values["enabled"] = result["enabled"] is True
|
|
if "restart_required" in result:
|
|
status_values["restart_required"] = result["restart_required"] is True
|
|
if status_values:
|
|
if status_values.get("restart_required"):
|
|
status_values["state"] = ManagedPlugin.State.RESTART_REQUIRED
|
|
elif status_values.get("installed_version"):
|
|
status_values["state"] = (
|
|
ManagedPlugin.State.ENABLED
|
|
if status_values.get("enabled")
|
|
else ManagedPlugin.State.DISABLED
|
|
)
|
|
else:
|
|
status_values["state"] = ManagedPlugin.State.UNKNOWN
|
|
status_values.update({"last_error": "", "last_checked": timezone.now()})
|
|
ManagedPlugin.objects.filter(slug=audit.slug).update(**status_values)
|
|
elif state in {"failed", "errored", "cancelled"}:
|
|
error = redact_text(body.get("error") or f"Host agent operation {state}.")
|
|
LifecycleAudit.objects.filter(pk=audit.pk).update(
|
|
status=LifecycleAudit.Status.FAILED,
|
|
result_data=redact_data(body),
|
|
error=error,
|
|
completed=timezone.now(),
|
|
)
|
|
ManagedPlugin.objects.filter(slug=audit.slug).update(
|
|
state=ManagedPlugin.State.FAILED,
|
|
last_error=error,
|
|
restart_required=True,
|
|
last_checked=timezone.now(),
|
|
)
|
|
elif state in {"queued", "pending", "running", "accepted"}:
|
|
LifecycleAudit.objects.filter(pk=audit.pk).update(result_data=redact_data(body))
|
|
except Exception as exc:
|
|
messages.warning(request, f"Host-Agent-Status konnte nicht aktualisiert werden: {redact_text(exc)}")
|