from __future__ import annotations import importlib.metadata import re import tempfile from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable, Protocol from urllib.parse import urlsplit, urlunsplit from uuid import uuid4 from packaging.version import InvalidVersion, Version from .agent import AgentClient from .client import CatalogPlugin, Release, StoreClient from .commands import SubprocessRunner from .editors import MutationReceipt, PluginConfigurationEditor, RequirementsEditor from .locking import FileLock from .redaction import redact_data, redact_text from .runtime import RuntimeSettings from .validation import ensure_not_self, validate_slug ACTIONS = ("install", "update", "enable", "disable", "uninstall") MUTATING_ACTIONS = frozenset(ACTIONS) class LifecycleError(RuntimeError): pass class LifecycleRepository(Protocol): def create_audit(self, request: "LifecycleRequest", actor_id: int | None) -> int: ... def mark_running(self, audit_id: int) -> None: ... def mark_success(self, audit_id: int, result: dict[str, Any]) -> None: ... def mark_handed_off(self, audit_id: int, operation_id: str, result: dict[str, Any]) -> None: ... def mark_failed(self, audit_id: int, error: str, result: dict[str, Any]) -> None: ... def update_status(self, plugin: CatalogPlugin, **values: Any) -> None: ... def has_other_pending_mutation(self, slug: str, audit_id: int) -> bool: ... @dataclass(frozen=True, slots=True) class LifecycleRequest: slug: str action: str version: str = "" dry_run: bool = True def validate(self) -> None: validate_slug(self.slug) if self.action not in ACTIONS: raise LifecycleError("Unknown lifecycle action.") if self.version: try: Version(self.version) except InvalidVersion as exc: raise LifecycleError("Invalid requested version.") from exc @dataclass(slots=True) class LifecycleResult: slug: str action: str state: str dry_run: bool installed_version: str = "" enabled: bool = False restart_required: bool = False handed_to_agent: bool = False operation_id: str = "" plan: list[str] = field(default_factory=list) output: str = "" def safe_dict(self) -> dict[str, Any]: return redact_data(asdict(self)) def installed_distribution_version(package_name: str) -> str: try: return importlib.metadata.version(package_name) except importlib.metadata.PackageNotFoundError: return "" def runtime_plugin_is_loaded(import_name: str) -> bool: try: from django.conf import settings as django_settings if not django_settings.configured: return False return import_name in django_settings.PLUGINS except (ImportError, AttributeError, RuntimeError): return False class LifecycleService: def __init__( self, settings: RuntimeSettings, client: StoreClient, repository: LifecycleRepository, *, runner: SubprocessRunner | None = None, version_provider: Callable[[str], str] = installed_distribution_version, runtime_active_provider: Callable[[str], bool] = runtime_plugin_is_loaded, ): self.settings = settings self.client = client self.repository = repository self.runner = runner or SubprocessRunner() self.version_provider = version_provider self.runtime_active_provider = runtime_active_provider def execute( self, request: LifecycleRequest, *, actor_id: int | None = None, audit_id: int | None = None, ) -> LifecycleResult: request.validate() if audit_id is None: audit_id = self.repository.create_audit(request, actor_id) self.repository.mark_running(audit_id) result: LifecycleResult | None = None try: if request.dry_run: result = self._execute_locked(request, mutate=False, requested_by=actor_id) else: if self.settings.execution_mode == "dry_run": raise LifecycleError( "Real lifecycle changes are disabled while execution_mode is dry_run." ) if self.settings.execution_mode == "direct" and not self.settings.allow_lifecycle_mutations: raise LifecycleError( "Direct lifecycle changes require the explicit allow_lifecycle_mutations opt-in." ) with FileLock(self.settings.lock_path, self.settings.lock_timeout): if self.repository.has_other_pending_mutation(request.slug, audit_id): raise LifecycleError("Another lifecycle operation for this plugin is still pending.") result = self._execute_locked(request, mutate=True, requested_by=actor_id) if result.handed_to_agent: self.repository.mark_handed_off(audit_id, result.operation_id, result.safe_dict()) else: self.repository.mark_success(audit_id, result.safe_dict()) return result except Exception as exc: safe_error = redact_text(exc) safe_result = result.safe_dict() if result else {"slug": request.slug, "action": request.action} self.repository.mark_failed(audit_id, safe_error, safe_result) raise def _execute_locked( self, request: LifecycleRequest, *, mutate: bool, requested_by: int | None ) -> LifecycleResult: plugin = self.client.get_plugin(request.slug) ensure_not_self(plugin.package_name, plugin.import_name) installed = self.version_provider(plugin.package_name) runtime_active = self.runtime_active_provider(plugin.import_name) config_editor: PluginConfigurationEditor | None = None requirements_editor: RequirementsEditor | None = None if self.settings.execution_mode == "agent": enabled = runtime_active else: config_editor = PluginConfigurationEditor( self.settings.configuration_path, self.settings.backup_dir, self.settings.backups_to_keep, ) requirements_editor = RequirementsEditor( self.settings.requirements_path, self.settings.backup_dir, self.settings.backups_to_keep, ) enabled = plugin.import_name in config_editor.enabled_plugins() release: Release | None = None if request.action in {"install", "update"}: release = plugin.select_release(self.settings.netbox_version, request.version) self._assert_installable(plugin, release) self._assert_state(request.action, installed, enabled, runtime_active) plan = self._build_plan(request.action, plugin, release, enabled) if not mutate: return LifecycleResult( slug=plugin.slug, action=request.action, state="dry-run", dry_run=True, installed_version=installed, enabled=enabled, restart_required=( request.action in {"enable", "disable"} or (request.action == "update" and enabled) ), plan=plan, ) self.repository.update_status( plugin, state="running", installed_version=installed, enabled=enabled, restart_required=False, last_error="", ) try: if self.settings.execution_mode == "agent": result = self._execute_agent( request, plugin, release, installed, enabled, plan, requested_by ) else: assert config_editor is not None and requirements_editor is not None result = self._execute_direct( request, plugin, release, installed, enabled, plan, config_editor, requirements_editor, ) except Exception as exc: self.repository.update_status( plugin, state="failed", installed_version=self.version_provider(plugin.package_name) or installed, enabled=enabled, restart_required=True, last_error=redact_text(exc, limit=4_000), ) raise self.repository.update_status( plugin, state=result.state, installed_version=result.installed_version, enabled=result.enabled, restart_required=result.restart_required, last_error="", ) return result @staticmethod def _assert_installable(plugin: CatalogPlugin, release: Release) -> None: if not plugin.approved: raise LifecycleError("Plugin is not explicitly approved by the Store.") if not release.approved or not release.immutable: raise LifecycleError("Release is not both approved and immutable.") if not release.download_url: raise LifecycleError("Approved release has no artifact URL.") if not release.sha256: raise LifecycleError("Approved release has no artifact SHA-256. Commit hashes are not accepted.") if release.artifact_kind not in {"wheel", "source_archive"}: raise LifecycleError("Approved release has an unsupported artifact type.") @staticmethod def _assert_state(action: str, installed: str, enabled: bool, runtime_active: bool) -> None: if action == "install" and installed: raise LifecycleError("Plugin is already installed; use update.") if action in {"update", "enable", "disable", "uninstall"} and not installed: raise LifecycleError("Plugin is not installed.") if action == "enable" and enabled: raise LifecycleError("Plugin is already enabled.") if action == "disable" and not enabled: raise LifecycleError("Plugin is already disabled.") if action == "uninstall" and (enabled or runtime_active): raise LifecycleError("Disable the plugin and restart NetBox before uninstalling it.") def _build_plan( self, action: str, plugin: CatalogPlugin, release: Release | None, currently_enabled: bool ) -> list[str]: plan: list[str] = [] if release: plan.extend( [ f"Download immutable release {release.version} from an allowlisted origin.", f"Verify artifact SHA-256 {release.sha256} before executing pip.", ] ) if action == "install": plan.extend(["Install the verified local artifact with pip.", "Pin local_requirements.txt atomically."]) plan.append("Keep the newly installed plugin disabled until an explicit enable action.") elif action == "update": plan.extend(["Update the persistent requirement pin atomically.", "Upgrade from the verified local artifact."]) elif action == "enable": target = ( "agent-managed plugin list" if self.settings.execution_mode == "agent" else "static PLUGINS list" ) plan.append(f"Add {plugin.import_name} to the {target} atomically.") elif action == "disable": target = ( "agent-managed plugin list" if self.settings.execution_mode == "agent" else "static PLUGINS list" ) plan.append(f"Remove {plugin.import_name} from the {target} atomically.") elif action == "uninstall": plan.extend(["Remove the persistent requirement pin atomically.", "Uninstall the distribution with pip."]) if action == "enable" or (action == "update" and currently_enabled): if self.settings.run_migrations: plan.append("Run NetBox database migrations without interaction.") if self.settings.collect_static: plan.append("Collect NetBox static files without interaction.") if action in {"enable", "disable"} or (action == "update" and currently_enabled): plan.append("Require a NetBox/web and worker restart before the new state is fully active.") return plan def _artifact_suffix(self, release: Release) -> str: filename = Path(urlsplit(release.download_url).path).name suffixes = Path(filename).suffixes[-2:] suffix = "".join(suffixes) or ".artifact" return suffix if re.fullmatch(r"\.[A-Za-z0-9.]{1,16}", suffix) else ".artifact" def _download(self, release: Release, directory: Path) -> Path: if release.artifact_kind == "source_archive": raise LifecycleError("Source archives require execution_mode='agent' for a verified local wheel build.") artifact = directory / f"artifact{self._artifact_suffix(release)}" self.client.download_artifact( release.download_url, artifact, expected_sha256=release.sha256, timeout=self.settings.download_timeout, max_bytes=self.settings.max_download_bytes, ) return artifact def _execute_agent( self, request: LifecycleRequest, plugin: CatalogPlugin, release: Release | None, installed: str, enabled: bool, plan: list[str], requested_by: int | None, ) -> LifecycleResult: if self.settings.agent_socket_path is None: raise LifecycleError("Host agent socket is not configured.") if release is not None and not release.approved_payload_sha256: raise LifecycleError("Store release is missing its opaque approved_payload_sha256 marker.") agent = AgentClient(self.settings.agent_socket_path, timeout=self.settings.agent_timeout) operation_id = agent.submit_operation( request_id=str(uuid4()), action=request.action, plugin_slug=plugin.slug, version=release.version if release else (request.version or None), approved_payload_sha256=release.approved_payload_sha256 if release else None, requested_by=f"netbox-user:{requested_by}" if requested_by is not None else "netbox-system", ) return LifecycleResult( slug=plugin.slug, action=request.action, state="handed-off", dry_run=False, installed_version=installed, enabled=enabled, restart_required=( request.action in {"enable", "disable"} or (request.action == "update" and enabled) ), handed_to_agent=True, operation_id=str(operation_id), plan=plan, output="Operation accepted by the host agent.", ) def _execute_direct( self, request: LifecycleRequest, plugin: CatalogPlugin, release: Release | None, installed: str, enabled: bool, plan: list[str], config_editor: PluginConfigurationEditor, requirements_editor: RequirementsEditor, ) -> LifecycleResult: output: list[str] = [] with tempfile.TemporaryDirectory(prefix="netbox-plugin-store-") as temp_name: artifact = self._download(release, Path(temp_name)) if release else None if request.action == "install": new_version, new_enabled = self._direct_install( plugin, release, artifact, requirements_editor, output ) elif request.action == "update": new_version, new_enabled = self._direct_update( plugin, release, artifact, enabled, requirements_editor, output ) elif request.action == "enable": self._direct_enable(plugin, config_editor, output) new_version, new_enabled = installed, True elif request.action == "disable": config_editor.set_enabled(plugin.import_name, False) new_version, new_enabled = installed, False else: self._direct_uninstall(plugin, enabled, config_editor, requirements_editor, output) new_version, new_enabled = "", False return LifecycleResult( slug=plugin.slug, action=request.action, state=( "restart-required" if request.action in {"enable", "disable"} or (request.action == "update" and enabled) else ("installed" if new_version else "unknown") ), dry_run=False, installed_version=new_version, enabled=new_enabled, restart_required=( request.action in {"enable", "disable"} or (request.action == "update" and enabled) ), plan=plan, output=redact_text("\n".join(output)), ) def _pip_install(self, artifact: Path, *, upgrade: bool) -> str: argv = [ self.settings.python_executable, "-m", "pip", "install", "--disable-pip-version-check", "--no-input", *self.settings.pip_extra_args, ] if upgrade: argv.append("--upgrade") if not self.settings.allow_package_index: argv.append("--no-index") argv.append(str(artifact)) return self.runner.run(argv, timeout=self.settings.operation_timeout).output def _pip_uninstall(self, package_name: str) -> str: argv = [self.settings.python_executable, "-m", "pip", "uninstall", "--yes", package_name] return self.runner.run(argv, timeout=self.settings.operation_timeout).output def _pip_check(self) -> str: argv = [self.settings.python_executable, "-m", "pip", "check"] return self.runner.run(argv, timeout=self.settings.operation_timeout).output @staticmethod def _requirement_line(plugin: CatalogPlugin, release: Release) -> str: parsed = urlsplit(release.download_url) fragment = f"sha256={release.sha256}" if parsed.fragment: fragment = parsed.fragment + "&" + fragment pinned_url = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, parsed.query, fragment)) return f"{plugin.package_name} @ {pinned_url}" def _manage(self, command: str) -> str: if not self.settings.manage_path.is_file(): raise LifecycleError("NetBox manage.py path does not exist; configure manage_path.") argv = [self.settings.python_executable, str(self.settings.manage_path), command, "--no-input"] return self.runner.run( argv, timeout=self.settings.operation_timeout, cwd=self.settings.manage_path.parent ).output def _post_enable_steps(self, output: list[str]) -> None: if self.settings.run_migrations: output.append(self._manage("migrate")) if self.settings.collect_static: output.append(self._manage("collectstatic")) def _direct_install( self, plugin: CatalogPlugin, release: Release, artifact: Path, requirements_editor: RequirementsEditor, output: list[str], ) -> tuple[str, bool]: requirement_receipt: MutationReceipt | None = None pip_installed = False try: output.append(self._pip_install(artifact, upgrade=False)) pip_installed = True output.append(self._pip_check()) if self.settings.manage_requirements_file: requirement_receipt = requirements_editor.set_requirement( plugin.package_name, self._requirement_line(plugin, release) ) except Exception: if requirement_receipt: requirements_editor.rollback(requirement_receipt) if pip_installed: try: self._pip_uninstall(plugin.package_name) except Exception: pass raise return release.version, False def _direct_update( self, plugin: CatalogPlugin, release: Release, artifact: Path, enabled: bool, requirements_editor: RequirementsEditor, output: list[str], ) -> tuple[str, bool]: receipt: MutationReceipt | None = None pip_updated = False try: if self.settings.manage_requirements_file: receipt = requirements_editor.set_requirement( plugin.package_name, self._requirement_line(plugin, release) ) output.append(self._pip_install(artifact, upgrade=True)) pip_updated = True output.append(self._pip_check()) if enabled: self._post_enable_steps(output) except Exception: if receipt and not pip_updated: requirements_editor.rollback(receipt) raise return release.version, enabled def _direct_enable( self, plugin: CatalogPlugin, config_editor: PluginConfigurationEditor, output: list[str] ) -> None: receipt = config_editor.set_enabled(plugin.import_name, True) try: self._post_enable_steps(output) except Exception: config_editor.rollback(receipt) raise def _direct_uninstall( self, plugin: CatalogPlugin, enabled: bool, config_editor: PluginConfigurationEditor, requirements_editor: RequirementsEditor, output: list[str], ) -> None: config_receipt: MutationReceipt | None = None requirement_receipt: MutationReceipt | None = None try: if self.settings.manage_requirements_file: requirement_receipt = requirements_editor.set_requirement(plugin.package_name, None) output.append(self._pip_uninstall(plugin.package_name)) except Exception: if config_receipt: config_editor.rollback(config_receipt) if requirement_receipt: requirements_editor.rollback(requirement_receipt) raise def build_service(*, repository: LifecycleRepository | None = None) -> LifecycleService: settings = RuntimeSettings.from_django() client = StoreClient( settings.store_url, settings.allowed_store_urls, settings.allowed_artifact_urls, api_token=settings.api_token, timeout=settings.request_timeout, ) if repository is None: from .repository import DjangoLifecycleRepository repository = DjangoLifecycleRepository() return LifecycleService(settings, client, repository)