256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Check and make the bounded first headless start of the pinned M9 installer VM."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
from typing import Sequence
|
|
import uuid
|
|
|
|
import guard
|
|
|
|
|
|
TARGET_UUID = "b14c850b-a21e-4a56-acca-785590f0c4ac"
|
|
TARGET_NAME = guard.TARGET_NAME
|
|
LAB_ROOT = guard.LAB_ROOT
|
|
TARGET_FOLDER = LAB_ROOT / TARGET_NAME
|
|
EXPECTED_VDI = TARGET_FOLDER / f"{TARGET_NAME}.vdi"
|
|
EXPECTED_VISO = TARGET_FOLDER / (
|
|
f"Unattended-{TARGET_UUID}-aux-iso.viso"
|
|
)
|
|
MAX_CAPTURE = guard.MAX_CAPTURE
|
|
|
|
|
|
class InstallBootError(Exception):
|
|
"""Pinned installer state failed validation or first start was uncertain."""
|
|
|
|
|
|
def _regular_owned_file(
|
|
path: Path, *, owner_uid: int, mode: int | None, label: str,
|
|
) -> Path:
|
|
try:
|
|
info = path.lstat()
|
|
except OSError as exc:
|
|
raise InstallBootError(f"{label} is unavailable") from exc
|
|
if (not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode)
|
|
or info.st_uid != owner_uid
|
|
or (mode is not None and stat.S_IMODE(info.st_mode) != mode)):
|
|
permission = "a regular current-user-owned file" if mode is None else (
|
|
f"current-user-owned mode {mode:04o}"
|
|
)
|
|
raise InstallBootError(f"{label} must be {permission}")
|
|
if info.st_nlink != 1:
|
|
raise InstallBootError(f"{label} must have exactly one hard link")
|
|
try:
|
|
return path.resolve(strict=True)
|
|
except OSError as exc:
|
|
raise InstallBootError(f"{label} is unavailable") from exc
|
|
|
|
|
|
def _check_installer_private_files(target_folder: Path, owner_uid: int) -> None:
|
|
"""Check installer material permissions without opening or displaying contents."""
|
|
def walk_error(error: OSError) -> None:
|
|
raise InstallBootError("private target folder cannot be inspected") from error
|
|
|
|
try:
|
|
walk = tuple(os.walk(target_folder, followlinks=False, onerror=walk_error))
|
|
except OSError as exc:
|
|
raise InstallBootError("private target folder cannot be inspected") from exc
|
|
for directory, subdirectories, filenames in walk:
|
|
directory_path = Path(directory)
|
|
if any((directory_path / name).is_symlink() for name in subdirectories):
|
|
raise InstallBootError("symlink in private target folder is prohibited")
|
|
for name in subdirectories:
|
|
child = directory_path / name
|
|
child_info = child.lstat()
|
|
if (not stat.S_ISDIR(child_info.st_mode) or child_info.st_uid != owner_uid
|
|
or stat.S_IMODE(child_info.st_mode) != 0o700):
|
|
raise InstallBootError("nested target directories must be current-user-owned mode 0700")
|
|
for filename in filenames:
|
|
entry = directory_path / filename
|
|
if entry.is_symlink():
|
|
raise InstallBootError("symlink in private target folder is prohibited")
|
|
# VirtualBox's unattended installer generates the sensitive auxiliary
|
|
# media and installer scripts here. Match names only; never read content.
|
|
if re.search(r"(?:viso|isolinux|preseed|postinstall)", entry.name, re.IGNORECASE):
|
|
_regular_owned_file(
|
|
entry, owner_uid=owner_uid, mode=0o600,
|
|
label="generated unattended installer file",
|
|
)
|
|
|
|
|
|
def _machine_info(*, lab_root: Path, runner: guard.Runner, owner_uid: int) -> tuple[str, dict[str, str]]:
|
|
"""Apply strict host identity checks while allowing only the known installer .viso."""
|
|
try:
|
|
guard._owned_private_directory(lab_root, owner_uid)
|
|
manifest = guard._read_manifest(lab_root / guard.IDENTITY_FILE, owner_uid)
|
|
selected_uuid = str(manifest["uuid"])
|
|
if selected_uuid != TARGET_UUID:
|
|
raise InstallBootError("identity manifest UUID is not the pinned installer target")
|
|
|
|
listed = guard._invoke(runner, ("VBoxManage", "list", "vms"))
|
|
rows: list[tuple[str, str]] = []
|
|
for line in listed.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
match = re.fullmatch(r'"([^"\r\n]+)" \{([0-9A-Fa-f-]{36})\}', line)
|
|
if match is None:
|
|
raise InstallBootError("malformed VirtualBox VM inventory")
|
|
name, raw_uuid = match.groups()
|
|
try:
|
|
vm_uuid = str(uuid.UUID(raw_uuid))
|
|
except ValueError as exc:
|
|
raise InstallBootError("malformed VirtualBox VM UUID") from exc
|
|
rows.append((name, vm_uuid))
|
|
if (len({name for name, _ in rows}) != len(rows)
|
|
or len({vm_uuid for _, vm_uuid in rows}) != len(rows)
|
|
or [(name, vm_uuid) for name, vm_uuid in rows
|
|
if name == TARGET_NAME or vm_uuid == TARGET_UUID]
|
|
!= [(TARGET_NAME, TARGET_UUID)]):
|
|
raise InstallBootError("pinned VM name and UUID are missing, duplicated, or mismatched")
|
|
|
|
info = guard._machine_values(guard._invoke(
|
|
runner, ("VBoxManage", "showvminfo", TARGET_UUID, "--machinereadable"),
|
|
))
|
|
except guard.GuardError as exc:
|
|
raise InstallBootError(f"strict host identity check failed: {exc}") from None
|
|
|
|
if info.get("name") != TARGET_NAME or info.get("UUID", "").lower() != TARGET_UUID:
|
|
raise InstallBootError("showvminfo identity does not match the pinned target")
|
|
marker = f"{TARGET_NAME}:{TARGET_UUID}"
|
|
if manifest.get("host_marker") != marker or info.get("description") != marker:
|
|
raise InstallBootError("manifest and VirtualBox host marker do not match")
|
|
if info.get("VMState") != "poweroff":
|
|
raise InstallBootError("pinned target must be powered off before first start")
|
|
|
|
folder = lab_root / TARGET_NAME
|
|
try:
|
|
folder_info = folder.lstat()
|
|
except OSError as exc:
|
|
raise InstallBootError("exact target folder is unavailable") from exc
|
|
if (not stat.S_ISDIR(folder_info.st_mode) or stat.S_ISLNK(folder_info.st_mode)
|
|
or folder_info.st_uid != owner_uid or stat.S_IMODE(folder_info.st_mode) != 0o700):
|
|
raise InstallBootError("exact target folder must be current-user-owned mode 0700")
|
|
resolved_root = lab_root.resolve(strict=True)
|
|
resolved_folder = folder.resolve(strict=True)
|
|
if resolved_folder != resolved_root / TARGET_NAME:
|
|
raise InstallBootError("exact target folder resolves outside the lab root")
|
|
|
|
cfg_file = Path(info.get("CfgFile", ""))
|
|
if cfg_file.is_symlink() or cfg_file.name != f"{TARGET_NAME}.vbox":
|
|
raise InstallBootError("VM configuration path is unexpected")
|
|
config = _regular_owned_file(
|
|
cfg_file, owner_uid=owner_uid, mode=None,
|
|
label="VM configuration file",
|
|
)
|
|
if config.parent != resolved_folder:
|
|
raise InstallBootError("VM configuration is outside the exact target folder")
|
|
|
|
expected_disk = _regular_owned_file(
|
|
EXPECTED_VDI, owner_uid=owner_uid, mode=None,
|
|
label="pinned target VDI",
|
|
)
|
|
expected_viso = _regular_owned_file(
|
|
EXPECTED_VISO, owner_uid=owner_uid, mode=0o600, label="generated auxiliary .viso",
|
|
)
|
|
if expected_disk.parent != resolved_folder or expected_viso.parent != resolved_folder:
|
|
raise InstallBootError("expected VM media resolves outside the exact target folder")
|
|
if info.get("storagecontrollername0") != "SATA" or any(
|
|
key.startswith("storagecontrollername") and key != "storagecontrollername0"
|
|
for key in info
|
|
):
|
|
raise InstallBootError("storage controller inventory is not the pinned SATA controller")
|
|
attachments = {
|
|
key: value for key, value in info.items()
|
|
if re.fullmatch(r"SATA-[0-9]+-[0-9]+", key)
|
|
}
|
|
if (attachments.get("SATA-0-0") != str(expected_disk)
|
|
or attachments.get("SATA-1-0") != str(expected_viso)
|
|
or any(attachments.get(f"SATA-{port}-0") != "none" for port in range(2, 30))
|
|
or set(attachments) != {f"SATA-{port}-0" for port in range(30)}):
|
|
raise InstallBootError("SATA media do not exactly match the pinned installer state")
|
|
for key, value in info.items():
|
|
if (key.startswith("SATA-") and not re.fullmatch(
|
|
r"SATA-(?:[0-9]+-[0-9]+|(?:ImageUUID|IsEjected|hot-pluggable|nonrotational|discard|tempeject)-[0-9]+-[0-9]+)", key)):
|
|
raise InstallBootError("unrecognized SATA attachment metadata")
|
|
if (not key.startswith("SATA-")
|
|
and re.fullmatch(r".+-[0-9]+-[0-9]+", key)):
|
|
raise InstallBootError("unexpected attached storage controller or media")
|
|
if re.fullmatch(r"SATA-[0-9]+-[0-9]+", key) and value not in {"none", str(expected_disk), str(expected_viso)}:
|
|
raise InstallBootError("unexpected attached media")
|
|
|
|
if {key for key in info if re.fullmatch(r"nic[0-9]+", key)} != {
|
|
f"nic{slot}" for slot in range(1, 9)
|
|
}:
|
|
raise InstallBootError("network adapter inventory is incomplete or has extra devices")
|
|
nics = {f"nic{slot}": info.get(f"nic{slot}") for slot in range(1, 9)}
|
|
if nics != {"nic1": "nat", **{f"nic{slot}": "none" for slot in range(2, 9)}}:
|
|
raise InstallBootError("network inventory is not the pinned NAT-only configuration")
|
|
if any(key.lower().startswith(("natpf", "forwarding(")) and value
|
|
for key, value in info.items()):
|
|
raise InstallBootError("NAT port forwarding is prohibited")
|
|
if any(key.startswith("SharedFolder") and value for key, value in info.items()):
|
|
raise InstallBootError("VirtualBox shared folders are prohibited")
|
|
|
|
_check_installer_private_files(resolved_folder, owner_uid)
|
|
return TARGET_UUID, info
|
|
|
|
|
|
def inspect_installer_state(
|
|
*, lab_root: Path = LAB_ROOT,
|
|
owner_uid: int | None = None,
|
|
runner: guard.Runner = guard.run_command,
|
|
) -> tuple[str, dict[str, str]]:
|
|
owner_uid = os.getuid() if owner_uid is None else owner_uid
|
|
return _machine_info(lab_root=lab_root, runner=runner, owner_uid=owner_uid)
|
|
|
|
|
|
def start_installer(
|
|
*, lab_root: Path = LAB_ROOT,
|
|
owner_uid: int | None = None,
|
|
runner: guard.Runner = guard.run_command,
|
|
) -> str:
|
|
target_uuid, _ = inspect_installer_state(
|
|
lab_root=lab_root, owner_uid=owner_uid, runner=runner,
|
|
)
|
|
try:
|
|
guard._invoke(runner, ("VBoxManage", "startvm", target_uuid, "--type=headless"))
|
|
except guard.GuardError:
|
|
raise InstallBootError(
|
|
"bounded first-start request failed or timed out; the VM may have started; "
|
|
"run read-only inspection before any further action"
|
|
) from None
|
|
return target_uuid
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("action", choices=("check", "start"))
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
if args.action == "check":
|
|
target_uuid, _ = inspect_installer_state()
|
|
else:
|
|
target_uuid = start_installer()
|
|
except (InstallBootError, guard.GuardError) as exc:
|
|
print(f"Installer-state guard refused: {exc}", file=sys.stderr)
|
|
return 2
|
|
if args.action == "check":
|
|
print(f"installer state checked for {TARGET_NAME} ({target_uuid}); VM remains powered off")
|
|
else:
|
|
print(
|
|
f"bounded first-start request completed for {TARGET_NAME} ({target_uuid}); "
|
|
"guest installation and boot are not verified"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|