mirror of
https://github.com/Strategic-Automation/violin.git
synced 2026-08-14 12:33:37 +02:00
96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""Runtime selection for native Kali/Parrot, Docker Kali, and local fallback."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
from collections.abc import Callable
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BackendResolution:
|
|
requested: str
|
|
resolved: str
|
|
platform: str
|
|
container: str = ""
|
|
mount: str = ""
|
|
fallback_reason: str = ""
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
return asdict(self)
|
|
|
|
|
|
def _native_kali_or_parrot(os_release: Path = Path("/etc/os-release")) -> bool:
|
|
if platform.system().lower() != "linux" or not os_release.is_file():
|
|
return False
|
|
text = os_release.read_text(encoding="utf-8", errors="replace").lower()
|
|
return "id=kali" in text or "id=parrot" in text or "kali linux" in text or "parrot os" in text
|
|
|
|
|
|
def _docker_container_ready(
|
|
container: str, engagement: Path, run: Callable = subprocess.run
|
|
) -> tuple[bool, str]:
|
|
if shutil.which("docker") is None:
|
|
return False, "docker executable not found"
|
|
try:
|
|
result = run(
|
|
["docker", "inspect", "--format", "{{.State.Running}}|{{json .Mounts}}", container],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return False, f"Docker container {container!r} probe timed out"
|
|
if result.returncode != 0:
|
|
return False, f"Docker container {container!r} is unavailable"
|
|
running, _, mounts = result.stdout.strip().partition("|")
|
|
expected = f"/engagements/{engagement.name}"
|
|
if running.strip().lower() != "true":
|
|
return False, f"Docker container {container!r} is not running"
|
|
if expected not in mounts:
|
|
return False, f"Docker container {container!r} does not mount {expected}"
|
|
return True, expected
|
|
|
|
|
|
def resolve_backend(
|
|
requested: str,
|
|
engagement: Path,
|
|
*,
|
|
container: str = "kali-pentest",
|
|
docker_probe: Callable = _docker_container_ready,
|
|
native_probe: Callable = _native_kali_or_parrot,
|
|
) -> BackendResolution:
|
|
"""Resolve a backend without installing packages or starting containers."""
|
|
choice = str(requested or "auto").strip().lower()
|
|
if choice not in {"auto", "local", "docker"}:
|
|
raise ValueError("backend must be auto, local, or docker")
|
|
host = platform.system().lower()
|
|
if choice == "local":
|
|
return BackendResolution(choice, "local", host)
|
|
if choice == "docker":
|
|
ready, detail = docker_probe(container, engagement)
|
|
if not ready:
|
|
raise ValueError(f"Docker backend unavailable: {detail}")
|
|
return BackendResolution(choice, "docker", host, container, detail)
|
|
if native_probe():
|
|
return BackendResolution(choice, "local", host)
|
|
ready, detail = docker_probe(container, engagement)
|
|
if ready:
|
|
return BackendResolution(choice, "docker", host, container, detail)
|
|
return BackendResolution(choice, "local", host, fallback_reason=detail)
|
|
|
|
|
|
def runtime_readiness(engagement: Path) -> dict[str, object]:
|
|
resolution = resolve_backend("auto", engagement)
|
|
return {
|
|
"native_kali_or_parrot": _native_kali_or_parrot(),
|
|
"docker_executable": bool(shutil.which("docker")),
|
|
"auto": resolution.to_dict(),
|
|
}
|