207 lines
9.5 KiB
Python
207 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class RuntimeConfigurationError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _netbox_root(django_settings: object) -> Path:
|
|
"""Return NETBOX_ROOT while supporting NetBox settings that expose paths as strings."""
|
|
base_dir = Path(getattr(django_settings, "BASE_DIR"))
|
|
return Path(getattr(django_settings, "NETBOX_ROOT", base_dir.parent))
|
|
|
|
|
|
def _argv_list(value: object, setting: str) -> tuple[tuple[str, ...], ...]:
|
|
if not isinstance(value, (list, tuple)):
|
|
raise RuntimeConfigurationError(f"{setting} must be a list of argv lists.")
|
|
commands: list[tuple[str, ...]] = []
|
|
for command in value:
|
|
if (
|
|
not isinstance(command, (list, tuple))
|
|
or not command
|
|
or any(not isinstance(arg, str) or "\x00" in arg for arg in command)
|
|
):
|
|
raise RuntimeConfigurationError(f"Each {setting} entry must be a non-empty argv list.")
|
|
commands.append(tuple(command))
|
|
return tuple(commands)
|
|
|
|
|
|
def _string_list(value: object, setting: str) -> tuple[str, ...]:
|
|
if not isinstance(value, (list, tuple)) or any(not isinstance(v, str) for v in value):
|
|
raise RuntimeConfigurationError(f"{setting} must be a list of strings.")
|
|
return tuple(value)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RuntimeSettings:
|
|
store_url: str
|
|
allowed_store_urls: tuple[str, ...]
|
|
allowed_artifact_urls: tuple[str, ...]
|
|
api_token: str
|
|
request_timeout: int
|
|
download_timeout: int
|
|
max_download_bytes: int
|
|
configuration_path: Path
|
|
requirements_path: Path
|
|
manage_requirements_file: bool
|
|
allow_lifecycle_mutations: bool
|
|
default_dry_run: bool
|
|
execution_mode: str
|
|
agent_socket_path: Path | None
|
|
agent_timeout: int
|
|
agent_poll_interval: float
|
|
manage_path: Path
|
|
lock_path: Path
|
|
backup_dir: Path
|
|
lock_timeout: int
|
|
operation_timeout: int
|
|
pip_extra_args: tuple[str, ...]
|
|
allow_package_index: bool
|
|
run_migrations: bool
|
|
collect_static: bool
|
|
background_jobs: bool
|
|
synchronous_fallback: bool
|
|
job_queue: str
|
|
auto_restart: bool
|
|
restart_commands: tuple[tuple[str, ...], ...]
|
|
restart_allowlist: tuple[tuple[str, ...], ...]
|
|
backups_to_keep: int
|
|
python_executable: str
|
|
netbox_version: str
|
|
|
|
@classmethod
|
|
def from_mapping(
|
|
cls,
|
|
raw: dict[str, Any],
|
|
*,
|
|
configuration_dir: str | os.PathLike[str],
|
|
netbox_root: str | os.PathLike[str],
|
|
base_dir: str | os.PathLike[str],
|
|
netbox_version: str,
|
|
) -> "RuntimeSettings":
|
|
configuration_dir = Path(configuration_dir)
|
|
netbox_root = Path(netbox_root)
|
|
base_dir = Path(base_dir)
|
|
|
|
config_path = Path(raw.get("configuration_path") or configuration_dir / "configuration.py")
|
|
requirements_path = Path(raw.get("requirements_path") or netbox_root / "local_requirements.txt")
|
|
|
|
configured_manage = raw.get("manage_path")
|
|
if configured_manage:
|
|
manage_path = Path(configured_manage)
|
|
else:
|
|
candidates = (base_dir / "manage.py", netbox_root / "netbox" / "manage.py", netbox_root / "manage.py")
|
|
manage_path = next((candidate for candidate in candidates if candidate.is_file()), candidates[0])
|
|
|
|
lock_path = Path(raw.get("lock_path") or configuration_dir / ".plugin-store.lock")
|
|
backup_dir = Path(raw.get("backup_dir") or configuration_dir / "plugin-store-backups")
|
|
restart_commands = _argv_list(raw.get("restart_commands", []), "restart_commands")
|
|
restart_allowlist = _argv_list(raw.get("restart_allowlist", []), "restart_allowlist")
|
|
|
|
if raw.get("auto_restart", False):
|
|
if not restart_commands:
|
|
raise RuntimeConfigurationError("auto_restart requires at least one restart command.")
|
|
denied = [command for command in restart_commands if command not in restart_allowlist]
|
|
if denied:
|
|
raise RuntimeConfigurationError("Every restart command must exactly match restart_allowlist.")
|
|
if str(raw.get("execution_mode", "dry_run")) == "direct":
|
|
raise RuntimeConfigurationError("auto_restart is forbidden in direct execution mode; use the host agent.")
|
|
|
|
pip_extra_args = _string_list(raw.get("pip_extra_args", []), "pip_extra_args")
|
|
if any("\x00" in arg for arg in pip_extra_args):
|
|
raise RuntimeConfigurationError("pip_extra_args contains an invalid NUL byte.")
|
|
|
|
settings = cls(
|
|
store_url=str(raw.get("store_url") or "").rstrip("/"),
|
|
allowed_store_urls=_string_list(raw.get("allowed_store_urls", []), "allowed_store_urls"),
|
|
allowed_artifact_urls=_string_list(raw.get("allowed_artifact_urls", []), "allowed_artifact_urls"),
|
|
api_token=str(raw.get("api_token", "")),
|
|
request_timeout=int(raw.get("request_timeout", 15)),
|
|
download_timeout=int(raw.get("download_timeout", 120)),
|
|
max_download_bytes=int(raw.get("max_download_bytes", 268_435_456)),
|
|
configuration_path=config_path,
|
|
requirements_path=requirements_path,
|
|
manage_requirements_file=bool(raw.get("manage_requirements_file", True)),
|
|
allow_lifecycle_mutations=bool(raw.get("allow_lifecycle_mutations", False)),
|
|
default_dry_run=bool(raw.get("default_dry_run", True)),
|
|
execution_mode=str(raw.get("execution_mode", "dry_run")),
|
|
agent_socket_path=Path(raw["agent_socket_path"]) if raw.get("agent_socket_path") else None,
|
|
agent_timeout=int(raw.get("agent_timeout", 30)),
|
|
agent_poll_interval=float(raw.get("agent_poll_interval", 1.0)),
|
|
manage_path=manage_path,
|
|
lock_path=lock_path,
|
|
backup_dir=backup_dir,
|
|
lock_timeout=int(raw.get("lock_timeout", 30)),
|
|
operation_timeout=int(raw.get("operation_timeout", 900)),
|
|
pip_extra_args=pip_extra_args,
|
|
allow_package_index=bool(raw.get("allow_package_index", False)),
|
|
run_migrations=bool(raw.get("run_migrations", True)),
|
|
collect_static=bool(raw.get("collect_static", True)),
|
|
background_jobs=bool(raw.get("background_jobs", True)),
|
|
synchronous_fallback=bool(raw.get("synchronous_fallback", True)),
|
|
job_queue=str(raw.get("job_queue", "default")),
|
|
auto_restart=bool(raw.get("auto_restart", False)),
|
|
restart_commands=restart_commands,
|
|
restart_allowlist=restart_allowlist,
|
|
backups_to_keep=int(raw.get("backups_to_keep", 25)),
|
|
python_executable=sys.executable,
|
|
netbox_version=str(netbox_version).split("-")[0],
|
|
)
|
|
settings.validate()
|
|
return settings
|
|
|
|
@classmethod
|
|
def from_django(cls) -> "RuntimeSettings":
|
|
from django.conf import settings as django_settings
|
|
|
|
plugin_settings = dict(django_settings.PLUGINS_CONFIG.get("netbox_plugin_store", {}))
|
|
config_cls = __import__("netbox_plugin_store", fromlist=["config"]).config
|
|
defaults = dict(config_cls.default_settings)
|
|
defaults.update(plugin_settings)
|
|
release = getattr(django_settings, "RELEASE", None)
|
|
version = getattr(release, "version", None) or django_settings.VERSION
|
|
base_dir = Path(django_settings.BASE_DIR)
|
|
return cls.from_mapping(
|
|
defaults,
|
|
configuration_dir=django_settings.CONFIGURATION_DIR,
|
|
netbox_root=_netbox_root(django_settings),
|
|
base_dir=base_dir,
|
|
netbox_version=version,
|
|
)
|
|
|
|
def validate(self) -> None:
|
|
from .client import URLPolicy
|
|
|
|
URLPolicy(self.allowed_store_urls).check(self.store_url)
|
|
if not self.allowed_artifact_urls:
|
|
raise RuntimeConfigurationError("allowed_artifact_urls cannot be empty.")
|
|
if self.execution_mode not in {"dry_run", "direct", "agent"}:
|
|
raise RuntimeConfigurationError("execution_mode must be 'dry_run', 'direct', or 'agent'.")
|
|
if self.execution_mode == "agent":
|
|
if self.agent_socket_path is None or not self.agent_socket_path.is_absolute():
|
|
raise RuntimeConfigurationError("agent mode requires an absolute agent_socket_path.")
|
|
if not 1 <= self.agent_timeout <= 300:
|
|
raise RuntimeConfigurationError("agent_timeout must be between 1 and 300 seconds.")
|
|
if not 0.1 <= self.agent_poll_interval <= 30:
|
|
raise RuntimeConfigurationError("agent_poll_interval must be between 0.1 and 30 seconds.")
|
|
if not 1 <= self.request_timeout <= 300:
|
|
raise RuntimeConfigurationError("request_timeout must be between 1 and 300 seconds.")
|
|
if not 1 <= self.download_timeout <= 3_600:
|
|
raise RuntimeConfigurationError("download_timeout must be between 1 and 3600 seconds.")
|
|
if not 1_024 <= self.max_download_bytes <= 2_147_483_648:
|
|
raise RuntimeConfigurationError("max_download_bytes is outside the accepted range.")
|
|
if not 1 <= self.operation_timeout <= 7_200:
|
|
raise RuntimeConfigurationError("operation_timeout must be between 1 and 7200 seconds.")
|
|
if not 1 <= self.lock_timeout <= 300:
|
|
raise RuntimeConfigurationError("lock_timeout must be between 1 and 300 seconds.")
|
|
if not 1 <= self.backups_to_keep <= 500:
|
|
raise RuntimeConfigurationError("backups_to_keep must be between 1 and 500.")
|
|
if not self.job_queue or any(char.isspace() for char in self.job_queue):
|
|
raise RuntimeConfigurationError("job_queue is invalid.")
|