kstyagi
d8a25209b9
- Introduced a new method `installed_kernel` in `ImageStore` to handle kernel verification and checksum validation. - Updated `Installer` to include additional packages: `nftables` and `conntrack`. - Refined `LifecycleService` to improve error handling and state management during VM creation, starting, stopping, and deletion. - Enhanced port management with new methods for setting and syncing ports, including validation for port rules. - Improved error handling in `StateStore` to raise specific `NotFoundError` and `ConflictError` exceptions. - Updated network management to ensure proper TAP device handling and firewall integration. - Enhanced API responses in `vms.py` to include policy status for better client-side handling. - Added validation for VM names and ports in `validation.py` to enforce stricter rules and prevent conflicts.
93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
"""Small, testable wrappers around required host-level operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Callable, Sequence
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .config import Settings
|
|
from .errors import CommandError, UvmError
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CommandResult:
|
|
args: tuple[str, ...]
|
|
returncode: int
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
|
|
|
|
class CommandRunner:
|
|
"""Execute host commands while keeping command construction testable."""
|
|
|
|
def __init__(self, emit: Callable[[str], None] | None = print) -> None:
|
|
self._emit = emit
|
|
|
|
def run(
|
|
self,
|
|
command: Sequence[str | Path],
|
|
*,
|
|
check: bool = True,
|
|
capture: bool = False,
|
|
input_text: str | None = None,
|
|
sensitive: bool = False,
|
|
timeout: float | None = None,
|
|
) -> CommandResult:
|
|
args = tuple(str(part) for part in command)
|
|
if self._emit is not None:
|
|
self._emit(f"+ {shlex.join(args)}")
|
|
|
|
try:
|
|
completed = subprocess.run(
|
|
args,
|
|
check=False,
|
|
text=True,
|
|
input=input_text,
|
|
stdout=subprocess.PIPE if capture else sys.stderr,
|
|
stderr=subprocess.PIPE if capture else None,
|
|
timeout=timeout,
|
|
)
|
|
except FileNotFoundError as error:
|
|
raise CommandError(f"required command was not found: {args[0]}") from error
|
|
except OSError as error:
|
|
raise CommandError(f"could not run {args[0]}: {error}") from error
|
|
except subprocess.TimeoutExpired as error:
|
|
raise CommandError(f"command timed out: {shlex.join(args)}") from error
|
|
|
|
result = CommandResult(
|
|
args=args,
|
|
returncode=completed.returncode,
|
|
stdout=completed.stdout or "",
|
|
stderr=completed.stderr or "",
|
|
)
|
|
if check and result.returncode != 0:
|
|
detail = "" if sensitive else result.stderr.strip() or result.stdout.strip()
|
|
suffix = f": {detail}" if detail else ""
|
|
raise CommandError(
|
|
f"command failed ({result.returncode}): {shlex.join(args)}{suffix}"
|
|
)
|
|
return result
|
|
|
|
|
|
def require_root() -> None:
|
|
if os.geteuid() != 0:
|
|
raise UvmError("this command needs root. Run it with sudo.")
|
|
|
|
|
|
def check_kvm() -> None:
|
|
kvm = Path("/dev/kvm")
|
|
if not kvm.exists():
|
|
raise UvmError("/dev/kvm does not exist. Enable hardware virtualization/KVM first.")
|
|
if not os.access(kvm, os.R_OK | os.W_OK):
|
|
raise UvmError("no read/write access to /dev/kvm. Run as root or grant KVM access.")
|
|
|
|
|
|
def ensure_data_directories(settings: Settings) -> None:
|
|
for path in (settings.base, settings.bin_dir, settings.images_dir, settings.vms_dir):
|
|
path.mkdir(parents=True, exist_ok=True)
|