211 lines
8.8 KiB
Python
211 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare the one pinned M9 VM for unattended installation; never start it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import selectors
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from typing import Callable, Sequence
|
|
|
|
import guard
|
|
|
|
|
|
TARGET_UUID = "b14c850b-a21e-4a56-acca-785590f0c4ac"
|
|
LAB_ROOT = guard.LAB_ROOT
|
|
ISO_PATH = LAB_ROOT / guard.EXPECTED_ISO
|
|
ISO_SHA256 = guard.EXPECTED_ISO_SHA256
|
|
ADDITIONS_ISO = Path("/usr/share/virtualbox/VBoxGuestAdditions.iso")
|
|
ADDITIONS_SHA256 = "4f51a073296de31cce53924860549149be5dc339f65dcd1dbf34fd7accefe8fb"
|
|
PASSWORD_DIR = Path("/tmp/wga-m9-private")
|
|
USER_PASSWORD_FILE = PASSWORD_DIR / "wga-password"
|
|
ROOT_PASSWORD_FILE = PASSWORD_DIR / "root-password"
|
|
MAX_CAPTURE = 1024 * 1024
|
|
TIMEOUT_SECONDS = 120
|
|
|
|
|
|
class PrepareError(Exception):
|
|
"""Preflight refused or unattended preparation stopped; output is redacted."""
|
|
|
|
|
|
Runner = Callable[[Sequence[str]], subprocess.CompletedProcess[str]]
|
|
|
|
|
|
def run_command(argv: Sequence[str]) -> subprocess.CompletedProcess[str]:
|
|
"""Run bounded VBoxManage while retaining its potentially secret output privately."""
|
|
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 PrepareError("VBoxManage output exceeded the preparation limit")
|
|
returncode = process.wait(timeout=max(0.0, deadline - time.monotonic()))
|
|
except (OSError, subprocess.TimeoutExpired, PrepareError):
|
|
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 _check_private_file(path: Path, owner_uid: int) -> None:
|
|
try:
|
|
info = path.lstat()
|
|
except OSError as exc:
|
|
raise PrepareError("private synthetic password file 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 stat.S_IMODE(info.st_mode) != 0o600
|
|
or info.st_size == 0):
|
|
raise PrepareError("synthetic password files must be nonempty current-user files with mode 0600")
|
|
|
|
|
|
def _check_private_credentials(owner_uid: int) -> None:
|
|
try:
|
|
info = PASSWORD_DIR.lstat()
|
|
except OSError as exc:
|
|
raise PrepareError("private synthetic password directory is unavailable") from exc
|
|
if (not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode)
|
|
or info.st_uid != owner_uid or stat.S_IMODE(info.st_mode) != 0o700):
|
|
raise PrepareError("synthetic password directory must be current-user-owned mode 0700")
|
|
_check_private_file(USER_PASSWORD_FILE, owner_uid)
|
|
_check_private_file(ROOT_PASSWORD_FILE, owner_uid)
|
|
|
|
|
|
def _check_digest(path: Path, expected: str, label: str) -> None:
|
|
try:
|
|
info = path.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode):
|
|
raise PrepareError(f"{label} must be a regular non-symlink file")
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for block in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
except OSError as exc:
|
|
raise PrepareError(f"{label} is unavailable") from exc
|
|
if digest.hexdigest() != expected:
|
|
raise PrepareError(f"{label} SHA-256 does not match the pinned value")
|
|
|
|
|
|
def _validate_preinstall_state(info: dict[str, str]) -> None:
|
|
if info.get("VMState") != "poweroff":
|
|
raise PrepareError("exact target must be powered off before preparation")
|
|
expected_disk = str((LAB_ROOT / guard.TARGET_NAME / f"{guard.TARGET_NAME}.vdi").resolve(strict=True))
|
|
expected_iso = str(ISO_PATH.resolve(strict=True))
|
|
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") != expected_disk
|
|
or attachments.get("SATA-1-0") != expected_iso):
|
|
raise PrepareError("target media do not match the expected VDI and pinned installer ISO")
|
|
extra_attachments = set(attachments) - {"SATA-0-0", "SATA-1-0"}
|
|
if any(attachments[key] not in {"none", "emptydrive"} for key in extra_attachments):
|
|
raise PrepareError("target media do not match the expected VDI and pinned installer ISO")
|
|
|
|
|
|
def _arguments(*, dry_run: bool) -> tuple[str, ...]:
|
|
args = [
|
|
"VBoxManage", "unattended", "install", TARGET_UUID,
|
|
f"--iso={ISO_PATH}",
|
|
"--user=wga", "--full-user-name=WireGuard Admin Lab",
|
|
f"--user-password-file={USER_PASSWORD_FILE}",
|
|
f"--admin-password-file={ROOT_PASSWORD_FILE}",
|
|
"--hostname=wga-m9-debian137.invalid", "--locale=en_US", "--time-zone=UTC",
|
|
"--country=US", "--package-selection-adjustment=minimal",
|
|
"--install-additions", f"--additions-iso={ADDITIONS_ISO}",
|
|
"--start-vm=none",
|
|
]
|
|
if dry_run:
|
|
args.append("--dry-run")
|
|
return tuple(args)
|
|
|
|
|
|
def _invoke(runner: Runner, argv: Sequence[str], stage: str) -> None:
|
|
try:
|
|
result = runner(argv)
|
|
except PrepareError as exc:
|
|
if str(exc) == "VBoxManage output exceeded the preparation limit":
|
|
raise PrepareError(f"{stage} output exceeded the preparation limit; output was suppressed") from None
|
|
raise PrepareError(f"{stage} failed or timed out; VBoxManage output was suppressed") from None
|
|
except Exception:
|
|
raise PrepareError(f"{stage} failed or timed out; VBoxManage output was suppressed") from None
|
|
if result.returncode != 0:
|
|
raise PrepareError(f"{stage} failed with status {result.returncode}; VBoxManage output was suppressed")
|
|
if len(result.stdout) > MAX_CAPTURE or len(result.stderr) > MAX_CAPTURE:
|
|
raise PrepareError(f"{stage} output exceeded the preparation limit; output was suppressed")
|
|
|
|
|
|
def prepare_target(*, runner: Runner = run_command, owner_uid: int | None = None) -> str:
|
|
owner_uid = os.getuid() if owner_uid is None else owner_uid
|
|
_check_private_credentials(owner_uid)
|
|
if guard.EXPECTED_ISO_SHA256 != ISO_SHA256:
|
|
raise PrepareError("configured installer digest differs from the pinned digest")
|
|
_check_digest(ISO_PATH, ISO_SHA256, "installer ISO")
|
|
_check_digest(ADDITIONS_ISO, ADDITIONS_SHA256, "Guest Additions ISO")
|
|
try:
|
|
target_uuid, info = guard.inspect_target(runner=runner)
|
|
except (guard.GuardError, OSError, subprocess.TimeoutExpired):
|
|
raise PrepareError("read-only exact-target guard failed; refusing preparation") from None
|
|
if target_uuid != TARGET_UUID:
|
|
raise PrepareError("read-only guard returned an unexpected target UUID")
|
|
_validate_preinstall_state(info)
|
|
_invoke(runner, _arguments(dry_run=True), "unattended dry run")
|
|
try:
|
|
_invoke(runner, _arguments(dry_run=False), "unattended preparation")
|
|
except PrepareError:
|
|
raise PrepareError(
|
|
f"unattended preparation stopped or partially completed for {TARGET_UUID}; "
|
|
"do not rerun; perform separate read-only inspection before any separately reviewed start"
|
|
) from None
|
|
return TARGET_UUID
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
target_uuid = prepare_target()
|
|
except PrepareError as exc:
|
|
print(f"Unattended preparation refused/stopped: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(
|
|
f"Unattended preparation completed for the pinned target ({target_uuid}); "
|
|
"VM remains unstarted. Perform separate read-only media inspection before any separately reviewed start."
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|