"""VirtualBox client abstraction built on top of ``VBoxManage``."""
from __future__ import annotations
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
from typing import Callable, Protocol, Sequence
from pysnap.core.access import SSH_RULE_NAME, parse_ssh_forwarding, ssh_forwarding_rule
from pysnap.core.models import (
ComPortSetting,
NicSetting,
VMHardware,
ImportCandidate,
SerialPortConfiguration,
VMInfo,
VMReference,
)
from pysnap.errors import CommandExecutionError, PySnapError
from pysnap.vbox.parsers import (
parse_extra_data,
parse_import_candidates,
parse_list_vms,
parse_machine_readable,
parse_snapshot_names,
split_groups,
)
[docs]
class RunnerProtocol(Protocol):
"""Describe the command runner used by the client."""
[docs]
def run(self, arguments: Sequence[str], timeout: float | None = None) -> str:
"""Execute a command and return its standard output.
:param arguments: Arguments passed to ``VBoxManage``.
:param timeout: Optional timeout in seconds.
:returns: Standard output.
"""
pass
[docs]
def run_streaming(
self,
arguments: Sequence[str],
output_callback: Callable[[str], None],
) -> str:
"""Execute a command and stream textual output chunks.
:param arguments: Arguments passed to ``VBoxManage``.
:param output_callback: Callback for output chunks.
:returns: Combined command output.
"""
pass
[docs]
class SubprocessRunner(RunnerProtocol):
"""Run ``VBoxManage`` commands using :mod:`subprocess`."""
MACOS_APP_BUNDLE_EXECUTABLE = Path(
"/Applications/VirtualBox.app/Contents/MacOS/VBoxManage"
)
WINDOWS_INSTALL_DIRECTORY_ENV_VARS = (
"VBOX_MSI_INSTALL_PATH",
"VBOX_INSTALL_PATH",
)
WINDOWS_DEFAULT_EXECUTABLES = (
r"C:\Program Files\Oracle\VirtualBox\VBoxManage.exe",
r"C:\Program Files (x86)\Oracle\VirtualBox\VBoxManage.exe",
)
def __init__(self, executable: str | None = None) -> None:
"""Initialize the subprocess runner.
The runner first checks an explicit executable path, then the
``VBOXMANAGE_EXECUTABLE`` environment variable, then ``PATH``.
On macOS it also falls back to the standard VirtualBox app-bundle path.
On Windows it checks the MSI install directory and the default
``Program Files`` locations.
:param executable: Optional VirtualBox command line executable.
"""
self.executable = self._resolve_executable(executable)
[docs]
def run(self, arguments: Sequence[str], timeout: float | None = None) -> str:
"""Execute ``VBoxManage`` and return standard output.
:param arguments: Arguments passed to ``VBoxManage``.
:param timeout: Optional timeout in seconds.
:returns: Standard output text.
:raises CommandExecutionError: If the command exits unsuccessfully or
does not finish before the timeout expires.
"""
command = [self.executable, *arguments]
try:
completed = subprocess.run(
command,
capture_output=True,
check=False,
text=True,
timeout=timeout,
)
except FileNotFoundError as error:
raise CommandExecutionError(command, "", str(error)) from error
except subprocess.TimeoutExpired as error:
raise CommandExecutionError(
command,
self._decode_captured_stream(error.stdout),
f"VBoxManage did not respond within {timeout} seconds.",
) from error
if completed.returncode != 0:
raise CommandExecutionError(command, completed.stdout, completed.stderr)
return completed.stdout
[docs]
def run_streaming(
self,
arguments: Sequence[str],
output_callback: Callable[[str], None],
) -> str:
"""Execute ``VBoxManage`` and stream merged output chunks.
:param arguments: Arguments passed to ``VBoxManage``.
:param output_callback: Callback for output chunks.
:returns: Combined command output.
:raises CommandExecutionError: If the command exits unsuccessfully.
"""
command = [self.executable, *arguments]
try:
process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
except FileNotFoundError as error:
raise CommandExecutionError(command, "", str(error)) from error
assert process.stdout is not None
chunks: list[str] = []
while True:
chunk = process.stdout.read(1)
if not chunk:
break
text = chunk.decode("utf-8", errors="replace")
chunks.append(text)
output_callback(text)
process.wait()
output = "".join(chunks)
if process.returncode != 0:
raise CommandExecutionError(command, output, "")
return output
def _resolve_executable(self, executable: str | None) -> str:
"""Resolve the executable path for ``VBoxManage``.
:param executable: Explicit executable override, if any.
:returns: Resolved executable path or command name.
"""
requested = executable or os.environ.get("VBOXMANAGE_EXECUTABLE") or "VBoxManage"
explicit_path = Path(requested).expanduser()
if self._looks_like_filesystem_path(requested) and explicit_path.is_file():
return str(explicit_path)
resolved = shutil.which(requested)
if resolved:
return resolved
if sys.platform == "win32":
for candidate in self._iter_windows_fallback_executables():
if Path(candidate).is_file():
return candidate
if sys.platform == "darwin" and self.MACOS_APP_BUNDLE_EXECUTABLE.is_file():
return str(self.MACOS_APP_BUNDLE_EXECUTABLE)
return requested
def _decode_captured_stream(self, captured: str | bytes | None) -> str:
"""Decode output captured before a subprocess timeout.
:param captured: Raw captured stream content, if any.
:returns: Text representation of the captured content.
"""
if captured is None:
return ""
if isinstance(captured, bytes):
return captured.decode("utf-8", errors="replace")
return captured
def _looks_like_filesystem_path(self, value: str) -> bool:
"""Return whether a command value looks like a filesystem path.
:param value: Requested command value.
:returns: ``True`` when the value should be checked as a path.
"""
separators = {os.sep}
if os.altsep:
separators.add(os.altsep)
return value.startswith("~") or any(separator in value for separator in separators)
def _iter_windows_fallback_executables(self) -> tuple[str, ...]:
"""Return standard Windows candidates for ``VBoxManage.exe``.
:returns: Candidate executable paths in resolution order.
"""
candidates: list[str] = []
for variable in self.WINDOWS_INSTALL_DIRECTORY_ENV_VARS:
install_location = os.environ.get(variable)
if not install_location:
continue
candidates.append(self._normalize_windows_vboxmanage_path(install_location))
candidates.extend(self.WINDOWS_DEFAULT_EXECUTABLES)
return tuple(dict.fromkeys(candidates))
def _normalize_windows_vboxmanage_path(self, install_location: str) -> str:
"""Normalize one Windows VirtualBox install location to ``VBoxManage.exe``.
:param install_location: Directory or executable path reported by the OS.
:returns: Full ``VBoxManage.exe`` path candidate.
"""
normalized = install_location.strip().strip('"')
if normalized.lower().endswith(".exe"):
return normalized
return normalized.rstrip("\\/") + r"\VBoxManage.exe"
[docs]
class VBoxManageClient:
"""Expose the subset of VirtualBox operations required by PySnap."""
IMPORT_PROGRESS_PATTERN = re.compile(r"(?P<percent>\d{1,3})%")
DMI_SYSTEM_VENDOR_KEY = "VBoxInternal/Devices/pcbios/0/Config/DmiSystemVendor"
DMI_SYSTEM_SKU_KEY = "VBoxInternal/Devices/pcbios/0/Config/DmiSystemSKU"
# ``VBoxManage list`` can block forever when the VirtualBox configuration
# directory is missing or corrupted while a stale ``VBoxSVC`` still runs,
# for example right after the machine and configuration folders were
# deleted. A bounded timeout turns that hang into a reportable error.
LIST_COMMAND_TIMEOUT = 15.0
COM_COUNT = 4
# Network adapters per chipset when ``list systemproperties`` gives none.
DEFAULT_NIC_COUNTS = {"piix3": 8, "ich9": 36}
NIC_COUNT_PATTERN = re.compile(r"^Maximum (\w+) Network Adapter count:\s*(\d+)", re.MULTILINE)
# Standard PC I/O base addresses and IRQs of COM1-COM4.
COM_RESOURCES = {
1: ("0x3F8", "4"),
2: ("0x2F8", "3"),
3: ("0x3E8", "4"),
4: ("0x2E8", "3"),
}
def __init__(self, runner: RunnerProtocol | None = None) -> None:
"""Initialize the VirtualBox client.
:param runner: Command runner implementation.
"""
self.runner = runner or SubprocessRunner()
self._nic_counts: dict[str, int] | None = None
[docs]
def list_vms(self) -> list[VMReference]:
"""List all registered VMs.
:returns: Parsed VM references.
"""
return parse_list_vms(
self.runner.run(["list", "vms"], timeout=self.LIST_COMMAND_TIMEOUT)
)
[docs]
def list_running_vms(self) -> list[VMReference]:
"""List all currently running VMs.
:returns: Parsed references for running VMs.
"""
return parse_list_vms(
self.runner.run(["list", "runningvms"], timeout=self.LIST_COMMAND_TIMEOUT)
)
[docs]
def get_vm_info(self, vm_name: str) -> VMInfo:
"""Read detailed information about a VM.
:param vm_name: VM name to inspect.
:returns: Detailed VM information.
"""
properties = self._get_vm_properties(vm_name)
metadata = {
key: value
for key, value in self.get_metadata(vm_name).items()
if key.startswith("pysnap/")
}
return VMInfo(
name=properties.get("name", vm_name),
uuid=properties.get("UUID", ""),
groups=split_groups(properties.get("groups", "")),
serial_port=self._parse_serial_port(properties),
vm_state=properties.get("VMState", "") or None,
parent_name=metadata.get("pysnap/parent") or None,
managed=metadata.get("pysnap/managed") == "true",
metadata=metadata,
serial_tcp_ports=tuple(
setting.port
for setting in self._parse_com_ports(properties)
if setting.mode == "tcpserver" and setting.port is not None
),
ssh_port=(self._parse_ssh_rule(properties) or (None, None))[1],
)
[docs]
def add_ssh_forwarding(self, vm_name: str, adapter: int, port: int, *, running: bool) -> None:
"""Forward a host port on 127.0.0.1 to port 22 of the guest.
:param vm_name: VM name.
:param adapter: Number of the NAT adapter.
:param port: Host port.
:param running: Use ``controlvm`` for a running or paused VM instead of
``modifyvm``.
"""
rule = ssh_forwarding_rule(port)
if running:
self.runner.run(["controlvm", vm_name, f"natpf{adapter}", rule])
else:
self.runner.run(["modifyvm", vm_name, f"--natpf{adapter}", rule])
[docs]
def get_ssh_forwarding(self, vm_name: str) -> tuple[int, int] | None:
"""Return the NAT adapter and host port of the PySnap ssh rule.
:param vm_name: VM name.
:returns: ``(adapter, port)`` or ``None`` without such a rule.
"""
return self._parse_ssh_rule(self._get_vm_properties(vm_name))
[docs]
def remove_ssh_forwarding(self, vm_name: str, adapter: int) -> None:
"""Delete the PySnap ssh rule of a stopped VM.
:param vm_name: VM name.
:param adapter: Number of the NAT adapter holding the rule.
"""
self.runner.run(["modifyvm", vm_name, f"--natpf{adapter}", "delete", SSH_RULE_NAME])
[docs]
def type_on_console(self, vm_name: str, text: str) -> None:
"""Type a line on the screen of a running VM and press Enter.
``Ctrl-U`` first clears a half-typed line. VirtualBox types through
US keyboard scan codes, so the guest needs a Latin layout.
:param vm_name: VM name.
:param text: ASCII text to type.
"""
self.runner.run(["controlvm", vm_name, "keyboardputscancode", "1d", "16", "96", "9d"])
self.runner.run(["controlvm", vm_name, "keyboardputstring", text])
self.runner.run(["controlvm", vm_name, "keyboardputscancode", "1c", "9c"])
[docs]
def get_vm_hardware(self, vm_name: str) -> VMHardware:
"""Read the serial ports and network adapters of a VM.
:param vm_name: VM name to inspect.
:returns: Settings of COM1-COM4 and of every network adapter the
chipset of the VM offers.
"""
properties = self._get_vm_properties(vm_name)
nics = []
for number in range(1, self._nic_count(properties.get("chipset", "")) + 1):
attachment = (properties.get(f"nic{number}", "") or "none").strip().lower()
network = properties.get(f"intnet{number}") if attachment == "intnet" else None
nics.append(NicSetting(attachment=attachment, network=network or None))
return VMHardware(com_ports=self._parse_com_ports(properties), nics=tuple(nics))
[docs]
def apply_vm_hardware(
self,
vm_name: str,
com_ports: dict[int, ComPortSetting],
nics: dict[int, NicSetting],
) -> None:
"""Change serial ports and network adapters with one ``modifyvm`` call.
:param vm_name: VM name.
:param com_ports: Port numbers mapped to their new settings.
:param nics: Adapter numbers mapped to their new settings.
"""
arguments: list[str] = ["modifyvm", vm_name]
for number, setting in sorted(com_ports.items()):
if setting.mode == "off":
arguments.extend([f"--uart{number}", "off"])
continue
address, irq = self.COM_RESOURCES[number]
arguments.extend([f"--uart{number}", address, irq, f"--uartmode{number}"])
if setting.mode == "tcpserver":
arguments.extend(["tcpserver", str(setting.port)])
else:
arguments.extend([setting.mode, setting.path or ""])
for number, setting in sorted(nics.items()):
if setting.attachment == "intnet":
arguments.extend(
[f"--nic{number}", "intnet", f"--intnet{number}", setting.network or ""]
)
else:
arguments.extend([f"--nic{number}", setting.attachment])
self.runner.run(arguments)
[docs]
def set_dmi_system_sku(self, vm_name: str, system_sku: str) -> None:
"""Replace the DMI system SKU of a VM.
:param vm_name: VM name.
:param system_sku: New DMI system SKU value.
"""
self.runner.run(["setextradata", vm_name, self.DMI_SYSTEM_SKU_KEY, system_sku])
[docs]
def get_serial_port_configuration(self, vm_name: str) -> SerialPortConfiguration:
"""Read the raw ``UART1`` configuration of a VM.
:param vm_name: VM name to inspect.
:returns: Parsed ``UART1`` configuration.
"""
properties = self._get_vm_properties(vm_name)
uart1 = (properties.get("uart1", "") or "").strip()
uartmode1 = (properties.get("uartmode1", "") or "").strip()
if not uart1 or uart1.lower() == "off":
return SerialPortConfiguration(enabled=False)
if not uartmode1:
return SerialPortConfiguration(enabled=True)
mode_parts = [part.strip() for part in uartmode1.split(",") if part.strip()]
if not mode_parts:
return SerialPortConfiguration(enabled=True)
mode = mode_parts[0].lower()
port = None
if mode == "tcpserver" and len(mode_parts) >= 2:
try:
port = int(mode_parts[1])
except ValueError:
port = None
return SerialPortConfiguration(enabled=True, mode=mode, port=port)
[docs]
def dry_run_import(self, image_path: str) -> list[ImportCandidate]:
"""Run an appliance import dry run.
:param image_path: Appliance path.
:returns: Import candidates discovered by VirtualBox.
"""
output = self.runner.run(["import", image_path, "--dry-run"])
return parse_import_candidates(output)
[docs]
def import_appliance(
self,
image_path: str,
candidates: list[ImportCandidate],
progress_callback: Callable[[int], None] | None = None,
) -> None:
"""Import an appliance using the supplied dry-run candidates.
:param image_path: Appliance path.
:param candidates: Normalized import candidates.
:param progress_callback: Optional import progress callback.
"""
arguments: list[str] = ["import", image_path]
for candidate in candidates:
arguments.extend(
[
"--vsys",
str(candidate.vsys_index),
"--vmname",
candidate.vm_name,
"--group",
candidate.group,
]
)
if candidate.requires_eula_accept:
arguments.extend(
["--vsys", str(candidate.vsys_index), "--eula", "accept"]
)
if progress_callback is None:
self.runner.run(arguments)
return
progress_callback(0)
last_percent = 0
buffer = ""
def handle_output(chunk: str) -> None:
nonlocal buffer, last_percent
buffer = (buffer + chunk)[-128:]
matches = list(self.IMPORT_PROGRESS_PATTERN.finditer(buffer))
if not matches:
return
percent = min(int(matches[-1].group("percent")), 100)
if percent > last_percent:
last_percent = percent
progress_callback(percent)
self.runner.run_streaming(arguments, handle_output)
if last_percent < 100:
progress_callback(100)
[docs]
def get_current_snapshot_name(self, vm_name: str) -> str | None:
"""Return a snapshot name suitable for linked cloning.
:param vm_name: VM name to inspect.
:returns: Snapshot name, if any.
"""
snapshots = parse_snapshot_names(self._get_vm_properties(vm_name))
return snapshots[0] if snapshots else None
[docs]
def take_snapshot(self, vm_name: str, snapshot_name: str) -> None:
"""Create a snapshot for a VM.
:param vm_name: VM name.
:param snapshot_name: Snapshot name to create.
"""
self.runner.run(["snapshot", vm_name, "take", snapshot_name])
[docs]
def clone_linked(
self, base_vm: str, clone_vm: str, group: str, snapshot_name: str
) -> None:
"""Create and register a linked clone.
:param base_vm: Source VM name.
:param clone_vm: Clone VM name.
:param group: Group assigned to the clone.
:param snapshot_name: Snapshot used as clone source.
"""
self.runner.run(
[
"clonevm",
base_vm,
"--name",
clone_vm,
"--groups",
group,
"--snapshot",
snapshot_name,
"--options",
"link",
"--register",
]
)
[docs]
def start_vm_headless(self, vm_name: str) -> None:
"""Start a VM in headless mode.
:param vm_name: VM name.
"""
self.runner.run(["startvm", vm_name, "--type=headless"])
[docs]
def stop_vm_acpi(self, vm_name: str) -> None:
"""Request a graceful ACPI shutdown for a running VM.
:param vm_name: VM name.
"""
self.runner.run(["controlvm", vm_name, "acpipowerbutton"])
[docs]
def pause_vm(self, vm_name: str) -> None:
"""Pause a running VM; it stays in memory.
:param vm_name: VM name.
"""
self.runner.run(["controlvm", vm_name, "pause"])
[docs]
def resume_vm(self, vm_name: str) -> None:
"""Resume a paused VM.
:param vm_name: VM name.
"""
self.runner.run(["controlvm", vm_name, "resume"])
[docs]
def get_vm_state(self, vm_name: str) -> str:
"""Return the current VirtualBox runtime state of a VM.
:param vm_name: VM name.
:returns: Raw VirtualBox state string.
"""
return (self._get_vm_properties(vm_name).get("VMState", "") or "").lower()
[docs]
def delete_vm(self, vm_name: str) -> None:
"""Delete and unregister a VM.
:param vm_name: VM name.
"""
self.runner.run(["unregistervm", vm_name, "--delete"])
def _get_vm_properties(self, vm_name: str) -> dict[str, str]:
"""Read machine-readable VM properties.
:param vm_name: VM name.
:returns: Parsed machine-readable properties.
"""
return parse_machine_readable(
self.runner.run(["showvminfo", vm_name, "--machinereadable"])
)
@staticmethod
def _parse_ssh_rule(properties: dict[str, str]) -> tuple[int, int] | None:
"""Return the NAT adapter and host port of the PySnap ssh rule.
``showvminfo --machinereadable`` prints the ``Forwarding(i)`` rules of
a NAT adapter after its ``natnetN`` line.
:param properties: Parsed machine-readable VM properties in output order.
:returns: ``(adapter, port)`` or ``None``.
"""
adapter = 1
for key, value in properties.items():
if key.startswith("natnet") and key[6:].isdigit():
adapter = int(key[6:])
elif key.startswith("Forwarding("):
port = parse_ssh_forwarding(value)
if port is not None:
return adapter, port
return None
def _nic_count(self, chipset: str) -> int:
"""Return how many network adapters a chipset offers.
:param chipset: Chipset of the VM, such as ``piix3`` or ``ich9``.
:returns: Adapter count reported by VirtualBox, or a known default.
"""
chipset = (chipset or "piix3").strip().lower()
if self._nic_counts is None:
try:
output = self.runner.run(
["list", "systemproperties"], timeout=self.LIST_COMMAND_TIMEOUT
)
except CommandExecutionError:
output = ""
self._nic_counts = {
name.lower(): int(count) for name, count in self.NIC_COUNT_PATTERN.findall(output)
}
return self._nic_counts.get(chipset) or self.DEFAULT_NIC_COUNTS.get(chipset, 8)
def _parse_com_ports(self, properties: dict[str, str]) -> tuple[ComPortSetting, ...]:
"""Parse COM1-COM4 from machine-readable VM properties.
:param properties: Parsed machine-readable VM properties.
:returns: Settings of COM1-COM4.
"""
settings = []
for number in range(1, self.COM_COUNT + 1):
uart = (properties.get(f"uart{number}", "") or "").strip()
mode_value = (properties.get(f"uartmode{number}", "") or "").strip()
if not uart or uart.lower() == "off":
settings.append(ComPortSetting(mode="off"))
continue
mode, _, argument = mode_value.partition(",")
mode = mode.strip().lower()
argument = argument.strip()
if mode == "tcpserver":
port = int(argument) if argument.isdigit() else None
settings.append(ComPortSetting(mode="tcpserver", port=port))
elif mode in {"server", "client"}:
settings.append(ComPortSetting(mode=mode, path=argument))
else:
settings.append(
ComPortSetting(mode=mode or "disconnected", detail=argument or None)
)
return tuple(settings)
def _parse_serial_port(self, properties: dict[str, str]) -> int | None:
"""Extract the TCP port configured for ``UART1``.
:param properties: Parsed machine-readable VM properties.
:returns: TCP port or ``None`` when not configured as a TCP server.
"""
uart1 = properties.get("uart1", "")
uartmode1 = properties.get("uartmode1", "")
if not uart1 or uart1.lower() == "off" or not uartmode1:
return None
mode_parts = [part.strip() for part in uartmode1.split(",") if part.strip()]
if len(mode_parts) < 2 or mode_parts[0].lower() != "tcpserver":
return None
try:
return int(mode_parts[1])
except ValueError:
return None