285 lines
12 KiB
Python
285 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Create and validate the one approved M9 VirtualBox VM; never start it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import selectors
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from typing import Callable, Sequence
|
|
import uuid
|
|
|
|
import guard
|
|
|
|
|
|
LAB_ROOT = guard.LAB_ROOT
|
|
ISO_PATH = LAB_ROOT / guard.EXPECTED_ISO
|
|
VM_NAME = guard.TARGET_NAME
|
|
OS_TYPE = "Debian13_64"
|
|
DISK_SIZE_MIB = 8192
|
|
MEMORY_MIB = 1024
|
|
CPUS = 1
|
|
MAX_CAPTURE = 1024 * 1024
|
|
TIMEOUT_SECONDS = 30
|
|
|
|
|
|
class CreateError(Exception):
|
|
"""Creation could not safely complete; partial target is retained."""
|
|
|
|
|
|
Runner = Callable[[Sequence[str]], subprocess.CompletedProcess[str]]
|
|
|
|
|
|
def run_command(argv: Sequence[str]) -> subprocess.CompletedProcess[str]:
|
|
process = subprocess.Popen(
|
|
list(argv), stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
env={"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LANG": "C"},
|
|
)
|
|
assert process.stdout is not None and process.stderr is not None
|
|
output = {process.stdout: bytearray(), process.stderr: bytearray()}
|
|
selector = selectors.DefaultSelector()
|
|
for stream in output:
|
|
os.set_blocking(stream.fileno(), False)
|
|
selector.register(stream, selectors.EVENT_READ)
|
|
deadline = time.monotonic() + TIMEOUT_SECONDS
|
|
try:
|
|
while selector.get_map():
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise subprocess.TimeoutExpired(argv, TIMEOUT_SECONDS)
|
|
for key, _ in selector.select(remaining):
|
|
stream = key.fileobj
|
|
chunk = os.read(stream.fileno(), min(65536, MAX_CAPTURE + 1 - len(output[stream])))
|
|
if not chunk:
|
|
selector.unregister(stream)
|
|
continue
|
|
output[stream].extend(chunk)
|
|
if len(output[stream]) > MAX_CAPTURE:
|
|
raise CreateError("VBoxManage output exceeded the creation limit")
|
|
returncode = process.wait(timeout=max(0.0, deadline - time.monotonic()))
|
|
except (OSError, subprocess.TimeoutExpired, CreateError):
|
|
process.kill()
|
|
try:
|
|
process.wait(timeout=1)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
raise
|
|
finally:
|
|
selector.close()
|
|
process.stdout.close()
|
|
process.stderr.close()
|
|
return subprocess.CompletedProcess(
|
|
argv, returncode,
|
|
bytes(output[process.stdout]).decode("utf-8", errors="replace"),
|
|
bytes(output[process.stderr]).decode("utf-8", errors="replace"),
|
|
)
|
|
|
|
|
|
def _call(runner: Runner, argv: Sequence[str], *, created_uuid: str | None = None) -> str:
|
|
try:
|
|
result = runner(tuple(argv))
|
|
except (OSError, subprocess.TimeoutExpired, CreateError) as exc:
|
|
suffix = f"; retained partial VM {VM_NAME} ({created_uuid})" if created_uuid else ""
|
|
raise CreateError(f"bounded VBoxManage call failed: {argv[1]}{suffix}") from exc
|
|
if len(result.stdout) > MAX_CAPTURE or len(result.stderr) > MAX_CAPTURE:
|
|
raise CreateError("VBoxManage output exceeded the creation limit")
|
|
if result.returncode != 0:
|
|
suffix = f"; retained partial VM {VM_NAME} ({created_uuid})" if created_uuid else ""
|
|
raise CreateError(f"VBoxManage {argv[1]} failed with status {result.returncode}{suffix}")
|
|
return result.stdout
|
|
|
|
|
|
def _check_lab(lab_root: Path, iso_path: Path, owner_uid: int) -> None:
|
|
try:
|
|
root_info = lab_root.lstat()
|
|
except OSError as exc:
|
|
raise CreateError("private lab root is unavailable") from exc
|
|
if (not stat.S_ISDIR(root_info.st_mode) or stat.S_ISLNK(root_info.st_mode)
|
|
or root_info.st_uid != owner_uid or stat.S_IMODE(root_info.st_mode) != 0o700):
|
|
raise CreateError("lab root must be a real current-user directory with mode 0700")
|
|
if iso_path.parent.resolve(strict=True) != lab_root.resolve(strict=True):
|
|
raise CreateError("installer ISO must be directly inside the exact lab root")
|
|
try:
|
|
iso_info = iso_path.lstat()
|
|
except OSError as exc:
|
|
raise CreateError("pinned installer ISO is unavailable") from exc
|
|
if not stat.S_ISREG(iso_info.st_mode) or stat.S_ISLNK(iso_info.st_mode):
|
|
raise CreateError("installer ISO must be a regular non-symlink file")
|
|
digest = hashlib.sha256()
|
|
try:
|
|
with iso_path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
except OSError as exc:
|
|
raise CreateError("pinned installer ISO cannot be read") from exc
|
|
if digest.hexdigest() != guard.EXPECTED_ISO_SHA256:
|
|
raise CreateError("installer ISO SHA-256 does not match the pinned digest")
|
|
for path, description in (
|
|
(lab_root / guard.IDENTITY_FILE, "target identity manifest"),
|
|
(lab_root / VM_NAME, "exact target VM path"),
|
|
):
|
|
try:
|
|
path.lstat()
|
|
except FileNotFoundError:
|
|
continue
|
|
except OSError as exc:
|
|
raise CreateError(f"could not inspect {description}; refusing creation") from exc
|
|
if description == "target identity manifest":
|
|
raise CreateError(f"{description} already exists; refusing overwrite")
|
|
raise CreateError(f"{description} already exists; refusing creation")
|
|
|
|
|
|
def _check_absent(runner: Runner) -> None:
|
|
output = _call(runner, ("VBoxManage", "list", "vms"))
|
|
names: set[str] = set()
|
|
uuids: set[str] = set()
|
|
for line in output.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
match = re.fullmatch(r'"([^"\r\n]+)" \{([0-9A-Fa-f-]{36})\}', line)
|
|
if not match:
|
|
raise CreateError("malformed VBoxManage inventory; refusing creation")
|
|
name, raw_uuid = match.groups()
|
|
try:
|
|
vm_uuid = str(uuid.UUID(raw_uuid))
|
|
except ValueError as exc:
|
|
raise CreateError("malformed VBoxManage inventory UUID; refusing creation") from exc
|
|
if name in names or vm_uuid in uuids:
|
|
raise CreateError("duplicate VBoxManage inventory identity; refusing creation")
|
|
names.add(name)
|
|
uuids.add(vm_uuid)
|
|
if VM_NAME in names:
|
|
raise CreateError("exact target name is already registered; refusing creation")
|
|
|
|
|
|
def _write_manifest(lab_root: Path, target_uuid: str, owner_uid: int) -> None:
|
|
marker = f"{VM_NAME}:{target_uuid}"
|
|
data = {
|
|
"schema": 1,
|
|
"name": VM_NAME,
|
|
"owner_uid": owner_uid,
|
|
"uuid": target_uuid,
|
|
"host_marker": marker,
|
|
}
|
|
path = lab_root / guard.IDENTITY_FILE
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
try:
|
|
fd = os.open(path, flags, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as destination:
|
|
os.fchmod(destination.fileno(), 0o600)
|
|
json.dump(data, destination, sort_keys=True)
|
|
destination.write("\n")
|
|
destination.flush()
|
|
os.fsync(destination.fileno())
|
|
except OSError as exc:
|
|
raise CreateError(f"could not write identity manifest; retained partial VM {VM_NAME} ({target_uuid})") from exc
|
|
|
|
|
|
def create_target(
|
|
*, lab_root: Path = LAB_ROOT,
|
|
iso_path: Path = ISO_PATH,
|
|
owner_uid: int | None = None,
|
|
runner: Runner = run_command,
|
|
) -> str:
|
|
"""Preflight, create/configure the exact VM, and run the read-only guard."""
|
|
owner_uid = os.getuid() if owner_uid is None else owner_uid
|
|
try:
|
|
_check_lab(lab_root, iso_path, owner_uid)
|
|
_check_absent(runner)
|
|
except OSError as exc:
|
|
raise CreateError("creation preflight could not inspect the lab root or ISO") from exc
|
|
|
|
create_output = _call(runner, (
|
|
"VBoxManage", "createvm", "--name", VM_NAME, "--ostype", OS_TYPE,
|
|
"--basefolder", str(lab_root), "--register",
|
|
))
|
|
created_match = re.search(r"(?m)^UUID:\s*([0-9A-Fa-f-]{36})\s*$", create_output)
|
|
target_uuid: str | None = None
|
|
if created_match:
|
|
try:
|
|
target_uuid = str(uuid.UUID(created_match.group(1)))
|
|
except ValueError:
|
|
target_uuid = None
|
|
try:
|
|
info_text = _call(runner, (
|
|
"VBoxManage", "showvminfo", target_uuid or VM_NAME, "--machinereadable",
|
|
), created_uuid=target_uuid)
|
|
info = guard._machine_values(info_text)
|
|
if info.get("name") != VM_NAME or "UUID" not in info:
|
|
raise CreateError(f"created VM identity is unreadable; retained partial VM {VM_NAME} ({target_uuid or 'UUID unavailable'})")
|
|
reported_uuid = str(uuid.UUID(info["UUID"]))
|
|
if target_uuid is not None and target_uuid != reported_uuid:
|
|
raise CreateError(f"created VM UUID changed unexpectedly; retained partial VM {VM_NAME} ({target_uuid})")
|
|
target_uuid = reported_uuid
|
|
expected_cfg = lab_root / VM_NAME / f"{VM_NAME}.vbox"
|
|
if Path(info.get("CfgFile", "")) != expected_cfg:
|
|
raise CreateError(f"created VM path is unexpected; retained partial VM {VM_NAME} ({target_uuid})")
|
|
|
|
_call(runner, (
|
|
"VBoxManage", "modifyvm", target_uuid,
|
|
"--memory", str(MEMORY_MIB), "--cpus", str(CPUS),
|
|
"--nic1", "nat", "--nic2", "none", "--nic3", "none", "--nic4", "none",
|
|
"--nic5", "none", "--nic6", "none", "--nic7", "none", "--nic8", "none",
|
|
"--audio", "none", "--usb", "off",
|
|
"--boot1", "dvd", "--boot2", "disk", "--boot3", "none", "--boot4", "none",
|
|
), created_uuid=target_uuid)
|
|
disk = lab_root / VM_NAME / f"{VM_NAME}.vdi"
|
|
_call(runner, (
|
|
"VBoxManage", "createmedium", "disk", "--filename", str(disk),
|
|
"--size", str(DISK_SIZE_MIB), "--format", "VDI", "--variant", "Standard",
|
|
), created_uuid=target_uuid)
|
|
_call(runner, (
|
|
"VBoxManage", "storagectl", target_uuid, "--name", "SATA",
|
|
"--add", "sata", "--controller", "IntelAhci",
|
|
), created_uuid=target_uuid)
|
|
_call(runner, (
|
|
"VBoxManage", "storageattach", target_uuid, "--storagectl", "SATA",
|
|
"--port", "0", "--device", "0", "--type", "hdd", "--medium", str(disk),
|
|
), created_uuid=target_uuid)
|
|
_call(runner, (
|
|
"VBoxManage", "storageattach", target_uuid, "--storagectl", "SATA",
|
|
"--port", "1", "--device", "0", "--type", "dvddrive", "--medium", str(iso_path),
|
|
), created_uuid=target_uuid)
|
|
_write_manifest(lab_root, target_uuid, owner_uid)
|
|
_call(runner, (
|
|
"VBoxManage", "modifyvm", target_uuid,
|
|
"--description", f"{VM_NAME}:{target_uuid}",
|
|
), created_uuid=target_uuid)
|
|
try:
|
|
guard.inspect_target(lab_root=lab_root, owner_uid=owner_uid, runner=runner)
|
|
except (guard.GuardError, OSError) as exc:
|
|
raise CreateError(f"final read-only guard failed; retained partial VM {VM_NAME} ({target_uuid})") from exc
|
|
return target_uuid
|
|
except CreateError as exc:
|
|
if "retained partial" in str(exc):
|
|
raise
|
|
raise CreateError(f"{exc}; retained partial VM {VM_NAME} ({locals().get('target_uuid', 'UUID unavailable')})") from exc
|
|
except (ValueError, KeyError, OSError) as exc:
|
|
raise CreateError(f"creation identity/configuration failed; retained partial VM {VM_NAME} ({locals().get('target_uuid', 'UUID unavailable')})") from exc
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.parse_args(argv)
|
|
try:
|
|
target_uuid = create_target()
|
|
except CreateError as exc:
|
|
print(f"VM creation refused/stopped: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(f"Created and guarded {VM_NAME} ({target_uuid}); VM remains powered off")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|