542 lines
22 KiB
Python
542 lines
22 KiB
Python
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)
|
|
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()
|
|
runtime_active = self.runtime_active_provider(plugin.import_name)
|
|
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:
|
|
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":
|
|
plan.append(f"Add {plugin.import_name} to the static PLUGINS list atomically.")
|
|
elif action == "disable":
|
|
plan.append(f"Remove {plugin.import_name} from the static PLUGINS list 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)
|