diff --git a/AGENTS.md b/AGENTS.md index 8e415b8..b3578f0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,6 +15,14 @@ Develop inspectable, reproducible, **defensive** reliability and security resear 5. Use an isolated guest network without bridged access by default; never assume virtualization alone prevents egress. Validate containment before every experimental campaign. 6. On unexpected access, sensitive data exposure, or scope drift: stop execution, preserve minimal evidence securely, and escalate to the owner. +## Local testbed gate + +- Never enable GitHub Pages, publish a project URL/dashboard, or add a hosted endpoint for this repository. +- Keep guest networking disabled by default. No NIC, NAT, bridge, public graphics or host passthrough without separate explicit scope review. +- Never run scripts/testbed.py prepare or snapshot, virsh define/start, or any other state-changing host step without operator approval. GitHub CI must only test pure/mocked code. +- Do not represent a successful static XML audit as proof of real network containment or a completed RHEL experiment. +- Stage Daybreak repository code reviews separately. No scans or automated fixes are authorized merely by creating this PR. + ## Engineering defaults - Prefer Python standard library; use uv run --no-project for standalone scripts. If dependency management is truly needed, use uv add and uv lock with explicit versions. diff --git a/README.md b/README.md index 51df169..a65b631 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ RHEL-LAB is an Apache-2.0-licensed research workspace for controlled experiments on RHEL installations in isolated virtual machines running on owned or explicitly authorized hardware. The project prioritizes exact provenance, independently repeatable results, and responsible disclosure. -> Status: RHEL-LAB-001 foundation. This repository does not contain a tested RHEL VM image, demonstrated vulnerability, or completed kernel-fuzzing campaign. +> Status: RHEL-LAB-002 virtual-testbed tooling proposed for review. No RHEL VM has been installed, tested or booted by this repository. ## Research boundaries @@ -25,6 +25,7 @@ See [AGENTS.md](AGENTS.md), [SECURITY.md](SECURITY.md), and [docs/RESEARCH_PROTO | docs/RESEARCH_PROTOCOL.md | Experiment gates, provenance, disclosure | | docs/ROADMAP.md | Phased milestones and acceptance criteria | | scripts/evidence.py | Dependency-free SHA-256 evidence inventory and verifier | +| scripts/testbed.py | Operator-gated local KVM profile, auditing and offline disk snapshots | | tests/ | Safe standard-library unit tests | | .github/workflows/ci.yml | Non-privileged source checks only | @@ -45,6 +46,20 @@ To inventory an evidence directory, write the manifest **outside** the evidence The example requires the capture directory to exist. The manifest is deterministic (no generated timestamp), records relative paths, byte sizes and SHA-256 digests, rejects symlinks, and will not overwrite an existing manifest. For provenance, separately record target/kernel/package versions, test configuration, seed, exact source SHA, time, tool versions, and whether the test actually executed. A manifest establishes byte identity, not scientific validity. +## Virtual testbed (RHEL-LAB-002) + +The optional KVM/QEMU testbed is **offline-first, local-only, and operator-gated**. It uses an unprivileged libvirt user connection, a VM profile with no virtual NIC, serial-console-only installation, and private disk/snapshot state outside the repository. It has no Pages deployment or public endpoint. + +- [Virtual Testbed guide](docs/VIRTUAL_TESTBED.md) — plan, manual preparation, definition/audit, offline snapshots, limitations. +- [Daybreak security gate](docs/DAYBREAK_GATE.md) — source scanning and fixes come **before** any authorized live experimentation. +- [Lab profile](lab/testbed.json) and [baseline receipt template](templates/baseline_receipt.json). + +Once a local, authorized installer ISO is available, inspect the zero-write plan with: + + uv run --no-project python scripts/testbed.py plan --lab-root "$HOME/rhel-lab-state" --iso "$HOME/Downloads/RHEL-INSTALLER.iso" + +This command only produces an XML/JSON plan and performs no host mutation. Host-changing commands require separate explicit approval; no actual guest, security scan or isolated network proof has been performed. + ## Research priorities 1. Establish isolated, disposable RHEL guests and capture exact environment baselines. diff --git a/docs/DAYBREAK_GATE.md b/docs/DAYBREAK_GATE.md new file mode 100644 index 0000000..7f7c005 --- /dev/null +++ b/docs/DAYBREAK_GATE.md @@ -0,0 +1,25 @@ +# Daybreak Defense Factory — staged security gate + +**Status: NOT RUN.** RHEL-LAB-002 code should be reviewed through an ordinary GitHub PR and the non-privileged Python CI tests *before* requesting a Daybreak repository scan. Daybreak has not been authorized to run a scan by this milestone. + +## Controlled progression + +1. Review all changes, particularly the local command boundaries, operator approval tokens, XML allowlist, disk paths, and snapshot race limitations. +2. Check CI for the exact PR head SHA. Unit tests prove only the Python utilities' tested behaviors, not RHEL kernel behavior. +3. Request a **repository source-code security scan** through Daybreak when the owner approves it; use the connected QSOLKCB/RHEL-LAB repository and inspect findings. +4. Fix validated source-code findings on a new feature branch, reproduce reported problems, add regression tests and repeat CI/reviews. No automatic merging. +5. Approve and merge only after review gates pass. +6. Separately prepare/verify a local KVM guest, isolation, permitted licensing, disk capacity and offline access. The owner explicitly authorizes any actual host operation. +7. Only after those gates approve guest startup, local reliability experiments or other active testing. No Red Hat service/third-party probing. + +**Important distinction:** Daybreak scans repository source code. It is not a substitute for inspecting real guest firewall rules, proving isolation, running kernel fuzzing or observing a live RHEL VM. No GitHub Actions job may start a guest or conduct privileged/security testing. + +## Evidence to retain + +- Exact base/head/merge SHA and PR review records +- Python CI run ID, completion state, test outcomes and logs +- Daybreak scan run ID and actual findings once requested/executed +- Individual fixes, test reproductions, negative results and owner approvals +- Separately stored private VM evidence, logs, release/kernel package identities, snapshot hashes and disclosure records + +Do not claim "Daybreak clean" until the exact scan has completed successfully for the exact commit and its findings have been reviewed. Unfinished scans or zero results from a failed run are inconclusive. diff --git a/docs/LAB_SETUP.md b/docs/LAB_SETUP.md index be4025d..cc5f274 100644 --- a/docs/LAB_SETUP.md +++ b/docs/LAB_SETUP.md @@ -1,6 +1,6 @@ # Local RHEL lab setup — design checklist -This document specifies containment prerequisites. It is not an installation script and no hardware or guest has been provisioned by this repository. +This document specifies containment prerequisites. The companion scripts/testbed.py offers opt-in local VM preparation and snapshot commands, but no hardware or guest has been provisioned by repository CI. See VIRTUAL_TESTBED.md and DAYBREAK_GATE.md. ## Operator prerequisites diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index ae8d119..81a7e16 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -6,14 +6,20 @@ - [x] Document lab containment, experiment protocol and security disclosure. - [x] Supply deterministic, standard-library SHA-256 evidence inventory and verification. - [x] Supply safe unit tests and CI without guest workloads. -- [ ] Owner review and merge; CI status to be verified separately. +- [x] Owner reviewed and merged RHEL-LAB-001; CI verified green on the merge commit. -## RHEL-LAB-002 — authorized guest baseline +## RHEL-LAB-002 — The Virtual Testbed -- [ ] Select exact RHEL release and validate subscription/installation terms. -- [ ] Document actual hypervisor and guest configurations. -- [ ] Record signed-off VM scope, offline networking checks, exact package/kernel baseline. -- [ ] Create reproducible snapshot and artifact receipts. +- [x] Provide a strictly offline-only x86_64 KVM profile and serial-console-only libvirt domain XML. +- [x] Add no-write plan, approved disk preparation, persistent/live XML inspection and guarded offline snapshot commands. +- [x] Add a baseline receipt template, documented containment checklist, regression tests and Daybreak scan gate. +- [ ] Owner review, CI validation and security audit of this PR. +- [ ] Select/download exact RHEL ISO under valid entitlement and verify media integrity. +- [ ] Confirm actual host KVM/libvirt capability and security settings; record authorized VM scope. +- [ ] Manually define/install the isolated guest after authorization; verify *actual* network and host isolation. +- [ ] Capture a real clean baseline snapshot with exact guest/kernel/package evidence. + +Source tooling completion is not equivalent to guest installation, execution, isolation proof or a completed Daybreak scan. ## RHEL-LAB-003 — bounded reliability campaign diff --git a/docs/VIRTUAL_TESTBED.md b/docs/VIRTUAL_TESTBED.md new file mode 100644 index 0000000..30f3b59 --- /dev/null +++ b/docs/VIRTUAL_TESTBED.md @@ -0,0 +1,94 @@ +# RHEL-LAB-002 — The Virtual Testbed + +This milestone supplies an offline-first *host-side implementation*, not a preinstalled RHEL guest. No VM, kernel experiment, Red Hat service or Daybreak scan is executed in GitHub Actions. The testbed runs only on an operator-owned Linux host, after explicit approval. + +## Design constraints + +- KVM/QEMU under the least-privileged user libvirt connection: qemu:///session. No sudo or system-wide daemon required by these scripts. +- One x86_64 guest, 2 vCPUs, 4096 MiB RAM, standalone 48 GiB QCOW2 virtual disk. Configuration is in lab/testbed.json. +- **No guest NIC, bridge, NAT, hostdev, host share, USB redirection, graphics listener or public service** in the generated domain XML. +- Serial PTY console only. Guest ISO must be obtained separately from authorized Red Hat distribution; no proprietary image is committed. +- Images, private receipts and XML live in an operator-selected private 0700 directory **outside the Git repository**. +- Every host-changing command is explicit, opt-in, and fail-closed. No automatic boot, package installation, VM definition, snapshot restore, or host network configuration. +- A definition/active XML audit rejects unexpected disks, device passthrough, guest networking and external graphics. **This is a configuration audit, not proof of effective egress isolation.** +- No GitHub Pages, public dashboard or network endpoint. + +## Prepare the host manually + +Use a personally controlled Linux host with KVM support and local libvirt/QEMU tools installed by the operator according to their OS instructions. You need virsh, qemu-img, a working per-user qemu:///session connection, access to /dev/kvm and sufficient disk space. Do not install packages automatically as part of CI or this project. + +Check the actual hardware, hypervisor, firmware support, host ownership, VM resource budget, filesystem permissions and guest media authenticity yourself. The scripted plan **does not** check that a physical host is isolated. + +Download the correct RHEL installer ISO using your own authorized account, verify its official checksum/signature, and keep it outside Git. No vendor access credential belongs in a script, PR or public log. + +## Commands (host only, not GitHub Actions) + +From the checked-out repository, set paths before any mutating action: + + LAB="$HOME/rhel-lab-state" + ISO="$HOME/Downloads/RHEL-INSTALLER.iso" + +Preview the generated domain XML and paths without execution. The ISO must actually exist: + + uv run --no-project python scripts/testbed.py plan --lab-root "$LAB" --iso "$ISO" + +After manually reviewing the XML and approving disk preparation, run: + + uv run --no-project python scripts/testbed.py prepare \ + --lab-root "$LAB" --iso "$ISO" \ + --approve I_AUTHORIZE_LOCAL_VM_PREPARATION + +This creates the private folder, a blank QCOW2 image and domain.xml only. It **does not** define, start, install or connect a guest. The exact approval phrase is deliberate. + +Inspect domain.xml and then, **only after explicit owner authorization for this VM**, manually define and boot it with the user libvirt connection: + + virsh -c qemu:///session define "$LAB/domain.xml" + virsh -c qemu:///session start rhel-lab-offline-01 + virsh -c qemu:///session console rhel-lab-offline-01 + +The console is serial-only; the selected RHEL installer must support or be configured for serial/text installation (for example, console=ttyS0 and inst.text via the ISO's own boot configuration). If your ISO does not support that boot path, stop and make a reviewed change rather than adding a public graphics listener or a network adapter. + +Audit both persistent and current domain XML (including possible live changes): + + uv run --no-project python scripts/testbed.py audit-defined --lab-root "$LAB" --iso "$ISO" + +Re-run this after installation, host/libvirt configuration changes and before each experiment. The audit does not prove firewall state, process isolation or that the guest cannot access host services via other channels. A no-NIC VM still needs review of storage, console, device and host controls. + +## Offline snapshot and receipt + +First shut down the guest cleanly and establish that **no other process is writing its disk**. This step is mandatory. A guest-state check alone cannot prevent a process from starting the VM while copying; coordinate exclusive ownership of the host during conversion. + + virsh -c qemu:///session shutdown rhel-lab-offline-01 + virsh -c qemu:///session domstate rhel-lab-offline-01 + +Only when it reports shut off and an approved exclusive offline window exists, take a copy of its standalone QCOW2 disk: + + SHA="$(git rev-parse HEAD)" + uv run --no-project python scripts/testbed.py snapshot \ + --lab-root "$LAB" --iso "$ISO" \ + --label baseline-001 --source-sha "$SHA" \ + --approve I_AUTHORIZE_OFFLINE_SNAPSHOT + +The snapshot command: +1. Re-audits the persistent and current libvirt definitions. +2. Checks virsh domstate reports exactly "shut off". +3. Checks for a standalone QCOW2 image, then invokes qemu-img convert (source to new private image). +4. Refuses existing destination/receipt names and uses a no-clobber hard-link for final publication. +5. Records SHA-256, exact source commit, byte count and UTC time in a local receipt. + +This is an **offline disk copy**, not a libvirt RAM-state snapshot and not a guaranteed atomic capture against hostile/concurrent writers. Never use qemu-img against a running VM disk. + +Verify the resulting evidence directory using scripts/evidence.py, storing its manifest outside the directory being sealed. Preserve guest kernel, release, package versions, original media checksum, VM settings, snapshot identifiers, command line, exit status and negative findings in a separate private baseline receipt. + +## What this milestone does not do + +The repository does not deploy an RHEL guest automatically, bypass subscription conditions, install a hypervisor, configure the host firewall, prove network egress isolation, take a real snapshot in CI, run kernel tests, or claim vendor acceptance of findings. Owner signoff and independently verified isolation are required before RHEL-LAB-003. + +## Test and review gates + + uv run --no-project python -m unittest discover -s tests -v + uv run --no-project python -m compileall -q scripts tests + +The unit tests simulate unsafe profile changes, network injection, path substitution, root/symlink attacks, missing approvals, running guest snapshot refusal and both libvirt XML views. They do not invoke real QEMU/virsh or use a VM. + +Read AGENTS.md, docs/RESEARCH_PROTOCOL.md, and docs/DAYBREAK_GATE.md before approving any live operation. diff --git a/lab/testbed.json b/lab/testbed.json new file mode 100644 index 0000000..41382cb --- /dev/null +++ b/lab/testbed.json @@ -0,0 +1,9 @@ +{ + "schema": "qsol-rhel-testbed/v1", + "name": "rhel-lab-offline-01", + "architecture": "x86_64", + "network": "none", + "vcpus": 2, + "memory_mib": 4096, + "disk_gib": 48 +} diff --git a/scripts/testbed.py b/scripts/testbed.py new file mode 100644 index 0000000..d25a0a4 --- /dev/null +++ b/scripts/testbed.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python3 +"""Local-only KVM/libvirt testbed. No guest boot, network, or sudo commands.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import secrets +import shutil +import stat +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from xml.etree import ElementTree as ET + +SCHEMA = "qsol-rhel-testbed/v1" +REPO = Path(__file__).resolve().parent.parent +URI = "qemu:///session" +PREPARE_TOKEN = "I_AUTHORIZE_LOCAL_VM_PREPARATION" +SNAPSHOT_TOKEN = "I_AUTHORIZE_OFFLINE_SNAPSHOT" +NAME = re.compile(r"rhel-lab-[a-z0-9][a-z0-9-]{0,30}\Z") +LABEL = re.compile(r"[a-z][a-z0-9-]{0,39}\Z") +SHA = re.compile(r"[0-9a-f]{40}\Z") + + +class TestbedError(ValueError): + """Refused unsafe or unsupported operation.""" + + +def _pairs(items: list[tuple[str, object]]) -> dict: + out = {} + for key, value in items: + if key in out: + raise TestbedError(f"duplicate JSON property: {key}") + out[key] = value + return out + + +def read_config(path: Path) -> dict: + with path.open(encoding="utf-8") as stream: + cfg = json.load(stream, object_pairs_hook=_pairs) + fields = {"schema", "name", "architecture", "network", "vcpus", + "memory_mib", "disk_gib"} + if not isinstance(cfg, dict) or set(cfg) != fields: + raise TestbedError("unsupported configuration schema/fields") + if cfg["schema"] != SCHEMA or cfg["architecture"] != "x86_64" or cfg["network"] != "none": + raise TestbedError("only x86_64 and network=none are allowed") + if not isinstance(cfg["name"], str) or NAME.fullmatch(cfg["name"]) is None: + raise TestbedError("invalid VM name") + for field, lower, upper in (("vcpus", 1, 32), ("memory_mib", 1024, 32768), + ("disk_gib", 16, 512)): + if type(cfg[field]) is not int or not lower <= cfg[field] <= upper: + raise TestbedError(f"{field} must be integer between {lower} and {upper}") + return cfg + + +def absolute_path(text: str) -> Path: + path = Path(text) + if not path.is_absolute() or ".." in path.parts: + raise TestbedError("all host paths must be absolute without ..") + for entry in (path, *path.parents): + if entry.is_symlink(): + raise TestbedError(f"symlink in host path: {entry}") + return path.resolve(strict=False) + + +def state_dir(text: str, *, must_exist: bool = False) -> Path: + root = absolute_path(text) + if root == REPO or REPO in root.parents: + raise TestbedError("state must live outside the Git repository") + if root.exists(): + if not root.is_dir(): + raise TestbedError("state path is not a directory") + info = root.stat() + if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) & 0o077: + raise TestbedError("state directory must be user-owned and mode 0700") + elif must_exist: + raise TestbedError("state directory not prepared") + elif not root.parent.is_dir(): + raise TestbedError("state directory parent must already exist") + return root + + +def iso_file(text: str) -> Path: + path = absolute_path(text) + if path.suffix.lower() != ".iso" or not path.is_file(): + raise TestbedError("ISO must be an existing regular .iso file") + return path + + +def paths(root: Path) -> tuple[Path, Path]: + return root / "disk.qcow2", root / "domain.xml" + + +def render_xml(cfg: dict, root: Path, iso: Path) -> str: + disk, _ = paths(root) + domain = ET.Element("domain", {"type": "kvm"}) + ET.SubElement(domain, "name").text = cfg["name"] + ET.SubElement(domain, "memory", {"unit": "MiB"}).text = str(cfg["memory_mib"]) + ET.SubElement(domain, "vcpu").text = str(cfg["vcpus"]) + os_node = ET.SubElement(domain, "os") + ET.SubElement(os_node, "type", {"arch": "x86_64"}).text = "hvm" + ET.SubElement(os_node, "boot", {"dev": "cdrom"}) + ET.SubElement(os_node, "boot", {"dev": "hd"}) + devices = ET.SubElement(domain, "devices") + for kind, filename, fmt, target, bus in ( + ("disk", disk, "qcow2", "vda", "virtio"), + ("cdrom", iso, "raw", "sda", "sata"), + ): + node = ET.SubElement(devices, "disk", {"type": "file", "device": kind}) + ET.SubElement(node, "driver", {"name": "qemu", "type": fmt}) + ET.SubElement(node, "source", {"file": str(filename)}) + ET.SubElement(node, "target", {"dev": target, "bus": bus}) + if kind == "cdrom": + ET.SubElement(node, "readonly") + serial = ET.SubElement(devices, "serial", {"type": "pty"}) + ET.SubElement(serial, "target", {"port": "0"}) + console = ET.SubElement(devices, "console", {"type": "pty"}) + ET.SubElement(console, "target", {"type": "serial", "port": "0"}) + ET.indent(domain) + xml = ET.tostring(domain, encoding="unicode") + "\n" + audit_xml(xml, cfg, root, iso) + return xml + + +def audit_xml(xml: str, cfg: dict, root: Path, iso: Path) -> None: + """Reject interfaces, passthrough, external listeners and unexpected disk paths.""" + domain = ET.fromstring(xml) + if domain.tag != "domain" or domain.get("type") != "kvm": + raise TestbedError("domain must use KVM") + if any("}" in node.tag for node in domain.iter()): + raise TestbedError("namespaced QEMU overrides or extensions are forbidden") + if domain.findtext("name") != cfg["name"]: + raise TestbedError("VM name differs from approved profile") + devices = domain.find("devices") + if devices is None: + raise TestbedError("missing VM devices") + allowed = {"disk", "serial", "console", "controller", "input", "video", + "memballoon", "emulator"} + unexpected = {d.tag for d in devices} - allowed + if unexpected: + raise TestbedError(f"unexpected devices: {sorted(unexpected)}") + forbidden = {"interface", "graphics", "hostdev", "filesystem", "redirdev", + "channel", "shmem", "smartcard", "tpm", "commandline"} + if any(node.tag in forbidden for node in domain.iter()): + raise TestbedError("network, passthrough, or command-line override present") + os_type = domain.find("os/type") + if os_type is None or os_type.get("arch") != "x86_64" or os_type.text != "hvm": + raise TestbedError("unexpected guest architecture or OS type") + vcpu = domain.findtext("vcpu") + if vcpu != str(cfg["vcpus"]): + raise TestbedError("vCPU count does not match the profile") + memory = domain.find("memory") + if memory is None: + raise TestbedError("missing memory configuration") + memory_bytes = cfg["memory_mib"] * 1024 * 1024 + units = {"MiB": 1024 * 1024, "KiB": 1024} + if memory.get("unit") not in units or int(memory.text or "0") * units[memory.get("unit")] != memory_bytes: + raise TestbedError("memory allocation differs from the profile") + for port in devices.findall("serial") + devices.findall("console"): + if port.get("type") != "pty": + raise TestbedError("only local PTY serial/console devices are allowed") + if any(child.tag not in {"source", "target", "alias", "address"} for child in port): + raise TestbedError("unexpected serial/console attachment") + for source in port.findall("source"): + if set(source.attrib) != {"path"} or not re.fullmatch(r"/dev/pts/[0-9]+", source.get("path", "")): + raise TestbedError("unexpected PTY source path") + for item in devices.findall("input"): + if (item.get("type") not in {"mouse", "tablet", "keyboard"} + or item.get("bus") not in {None, "ps2", "usb", "virtio"} + or item.find("source") is not None): + raise TestbedError("host input passthrough is forbidden") + for item in devices.findall("video"): + if item.find(".//acceleration[@accel3d='yes']") is not None: + raise TestbedError("3D host GPU acceleration is forbidden") + emulator = devices.find("emulator") + if emulator is not None and emulator.text not in { + "/usr/bin/qemu-system-x86_64", "/usr/libexec/qemu-kvm" + }: + raise TestbedError("unapproved QEMU emulator executable") + if any(b.get("dev") not in {"cdrom", "hd"} for b in domain.findall(".//boot")): + raise TestbedError("non-disk or network boot is forbidden") + expected = {"disk": (str(root / "disk.qcow2"), "qcow2"), + "cdrom": (str(iso), "raw")} + actual_disks = devices.findall("disk") + if len(actual_disks) != 2: + raise TestbedError("expected exactly disk and ISO") + for item in actual_disks: + kind = item.get("device") + if kind not in expected or item.get("type") != "file": + raise TestbedError("unknown disk/device source") + file_path, fmt = expected.pop(kind) + src, driver = item.find("source"), item.find("driver") + if src is None or src.get("file") != file_path: + raise TestbedError(f"unsafe {kind} source") + if driver is None or driver.get("type") != fmt or driver.get("name") != "qemu": + raise TestbedError(f"unsafe {kind} format") + if kind == "cdrom" and item.find("readonly") is None: + raise TestbedError("ISO must remain read-only") + if expected: + raise TestbedError("disk or ISO missing") + + +def _diagnostic(value: str | bytes | None) -> str: + """Retain captured negative-result text across subprocess error types.""" + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="backslashreplace") + return value + + +def local_command(args: list[str], timeout: int = 30) -> str: + try: + result = subprocess.run(args, check=True, timeout=timeout, + capture_output=True, text=True) + except subprocess.CalledProcessError as exc: + raise TestbedError( + f"{args[0]} returned exit status {exc.returncode}; " + f"captured stdout: {_diagnostic(exc.stdout)}; " + f"captured stderr: {_diagnostic(exc.stderr)}" + ) from exc + except subprocess.TimeoutExpired as exc: + raise TestbedError( + f"{args[0]} timed out after {exc.timeout}s; " + f"captured stdout: {_diagnostic(exc.stdout)}; " + f"captured stderr: {_diagnostic(exc.stderr)}" + ) from exc + return result.stdout.strip() + + +def need(command: str) -> None: + if shutil.which(command) is None: + raise TestbedError(f"missing local dependency: {command}") + + +def approve(value: str | None, expected: str) -> None: + if value != expected: + raise TestbedError(f"explicit --approve {expected} required") + + +def plan(cfg: dict, root: Path, iso: Path) -> dict: + disk, definition = paths(root) + return { + "status": "PLANNED_NOT_EXECUTED", "connection": URI, "network": "none", + "disk": str(disk), "definition": str(definition), "iso": str(iso), + "size_gib": cfg["disk_gib"], "xml": render_xml(cfg, root, iso), + } + + +def prepare(cfg: dict, root: Path, iso: Path, token: str | None) -> dict: + approve(token, PREPARE_TOKEN) + need("qemu-img") + disk, definition = paths(root) + xml = render_xml(cfg, root, iso) + if any(path.exists() or path.is_symlink() for path in (disk, definition)): + raise TestbedError("refusing to replace existing disk/XML") + if not root.exists(): + root.mkdir(mode=0o700) + old_umask = os.umask(0o077) + try: + local_command(["qemu-img", "create", "-f", "qcow2", + str(disk), f"{cfg['disk_gib']}G"], timeout=120) + fd = os.open(definition, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + stream.write(xml) + finally: + os.umask(old_umask) + return {"status": "PREPARED_NOT_DEFINED_OR_BOOTED", + "definition": str(definition), "disk": str(disk), + "manual_next": f"virsh -c {URI} define {definition}"} + + +def audit_defined(cfg: dict, root: Path, iso: Path) -> dict: + need("virsh") + # Verify stored XML AND current (possibly live/hotplugged) XML. + for options in (["--inactive"], []): + xml = local_command(["virsh", "-c", URI, "dumpxml", *options, cfg["name"]]) + audit_xml(xml, cfg, root, iso) + return {"status": "DOMAIN_XML_AUDITED", "domain": cfg["name"], + "limitation": "Does not prove host egress isolation or prevent races."} + + +def hash_file(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + size += len(block) + return digest.hexdigest(), size + + +def snapshot(cfg: dict, root: Path, iso: Path, label: str, sha: str, + token: str | None) -> dict: + approve(token, SNAPSHOT_TOKEN) + if LABEL.fullmatch(label) is None or SHA.fullmatch(sha) is None: + raise TestbedError("invalid snapshot label or exact Git commit SHA") + need("qemu-img") + disk, _ = paths(root) + if disk.is_symlink() or not disk.is_file(): + raise TestbedError("local disk absent or symlinked") + folder = root / "snapshots" + if folder.is_symlink() or (folder.exists() and not folder.is_dir()): + raise TestbedError("snapshots directory is unsafe") + dest = folder / f"{label}.qcow2" + receipt = folder / f"{label}.receipt.json" + if any(p.exists() or p.is_symlink() for p in (dest, receipt)): + raise TestbedError("snapshot or receipt already exists") + audit_defined(cfg, root, iso) + state = local_command(["virsh", "-c", URI, "domstate", cfg["name"]]) + if state != "shut off": + raise TestbedError(f"domain is not shut off: {state}") + metadata = json.loads(local_command(["qemu-img", "info", "--output=json", str(disk)])) + if metadata.get("format") != "qcow2" or metadata.get("backing-filename"): + raise TestbedError("only standalone QCOW2 images are supported") + if not folder.exists(): + folder.mkdir(mode=0o700) + temporary = folder / f".pending-{secrets.token_hex(16)}.qcow2" + if temporary.exists(): + raise TestbedError("temporary snapshot name collision") + # QEMU warns against processing disks being used by another process. + local_command(["qemu-img", "convert", "-f", "qcow2", "-O", "qcow2", + str(disk), str(temporary)], timeout=3600) + if temporary.is_symlink() or not temporary.is_file(): + raise TestbedError("conversion did not produce a regular disk image") + digest, size = hash_file(temporary) + os.link(temporary, dest) # atomic no-clobber publication + temporary.unlink() + report = { + "schema": "qsol-rhel-snapshot/v1", "domain": cfg["name"], + "label": label, "source_commit": sha, "sha256": digest, + "bytes": size, "captured_utc": datetime.now(timezone.utc).isoformat(), + "claim": "OFFLINE_DISK_COPY_ONLY", + "limitation": "No guarantee against another process starting/writing this VM.", + } + with receipt.open("x", encoding="utf-8") as stream: + json.dump(report, stream, indent=2, sort_keys=True) + stream.write("\n") + return {"status": "SNAPSHOT_CREATED", "image": str(dest), + "receipt": str(receipt), "sha256": digest} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subs = parser.add_subparsers(dest="action", required=True) + for name in ("plan", "prepare", "audit-defined", "snapshot"): + p = subs.add_parser(name) + p.add_argument("--config", type=Path, default=REPO / "lab/testbed.json") + p.add_argument("--lab-root", required=True) + p.add_argument("--iso", required=True) + if name in {"prepare", "snapshot"}: + p.add_argument("--approve") + if name == "snapshot": + p.add_argument("--label", required=True) + p.add_argument("--source-sha", required=True) + args = parser.parse_args(argv) + try: + cfg = read_config(args.config) + root = state_dir(args.lab_root, must_exist=args.action in {"audit-defined", "snapshot"}) + iso = iso_file(args.iso) + if args.action == "plan": + report = plan(cfg, root, iso) + elif args.action == "prepare": + report = prepare(cfg, root, iso, args.approve) + elif args.action == "audit-defined": + report = audit_defined(cfg, root, iso) + else: + report = snapshot(cfg, root, iso, args.label, args.source_sha, args.approve) + except (TestbedError, OSError, ValueError, subprocess.SubprocessError, ET.ParseError) as exc: + print(f"FAIL: {exc}", file=sys.stderr) + return 1 + print(json.dumps(report, sort_keys=True, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/templates/baseline_receipt.json b/templates/baseline_receipt.json new file mode 100644 index 0000000..209706c --- /dev/null +++ b/templates/baseline_receipt.json @@ -0,0 +1,24 @@ +{ + "schema": "qsol-rhel-baseline/v1", + "evidence_class": "PLANNED", + "authorization_reference": null, + "test_scope": null, + "host_configuration": null, + "libvirt_connection": "qemu:///session", + "vm_name": "rhel-lab-offline-01", + "iso_sha256": null, + "rhel_release": null, + "kernel_release": null, + "installed_package_versions": null, + "guest_network_verification": null, + "guest_snapshot_sha256": null, + "repository_commit_sha": null, + "operator": null, + "collected_utc": null, + "commands_and_exit_codes": [], + "observations": [], + "negative_results": [], + "limitations": [ + "Template only. No guest has been created, tested, or independently verified." + ] +} diff --git a/tests/test_testbed.py b/tests/test_testbed.py new file mode 100644 index 0000000..b42c0de --- /dev/null +++ b/tests/test_testbed.py @@ -0,0 +1,206 @@ +"""Safe, mocked testbed tests. No virsh, qemu-img, KVM, root or network use.""" +from __future__ import annotations + +import json +import os +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from scripts import testbed as t + + +class TestbedTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + self.root = self.base / "vm-state" + self.iso = self.base / "rhel.iso" + self.iso.write_bytes(b"fixture, not installation media") + self.cfg = { + "schema": t.SCHEMA, "name": "rhel-lab-offline-01", + "architecture": "x86_64", "network": "none", + "vcpus": 2, "memory_mib": 4096, "disk_gib": 48, + } + self.config_path = self.base / "config.json" + self.write_config() + + def write_config(self): + self.config_path.write_text(json.dumps(self.cfg), encoding="utf-8") + + def test_valid_profile(self): + self.assertEqual(t.read_config(self.config_path), self.cfg) + + def test_unknown_fields_fail_closed(self): + self.cfg["external_network"] = True + self.write_config() + with self.assertRaises(t.TestbedError): + t.read_config(self.config_path) + + def test_no_network_allowed(self): + self.cfg["network"] = "nat" + self.write_config() + with self.assertRaises(t.TestbedError): + t.read_config(self.config_path) + + def test_bool_not_accepted_as_integer(self): + self.cfg["vcpus"] = True + self.write_config() + with self.assertRaises(t.TestbedError): + t.read_config(self.config_path) + + def test_duplicate_json_property_rejected(self): + self.config_path.write_text('{"name":"a","name":"b"}') + with self.assertRaises(t.TestbedError): + t.read_config(self.config_path) + + def test_xml_is_serial_only(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + self.assertIn("rhel-lab-offline-01", xml) + self.assertNotIn("", "") + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_tcp_serial_listener_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace('', '') + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_input_passthrough_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace("", '') + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_unapproved_emulator_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace("", "/tmp/untrusted-qemu") + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_changed_resource_limits_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace("2", "32") + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_libvirt_kib_memory_normalization_allowed(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace('4096', + '4194304') + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_file_path_substitution_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace(str(self.root / "disk.qcow2"), "/tmp/other.qcow2") + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_qemu_extension_rejected(self): + xml = t.render_xml(self.cfg, self.root, self.iso) + xml = xml.replace("", + '') + with self.assertRaises(t.TestbedError): + t.audit_xml(xml, self.cfg, self.root, self.iso) + + def test_symlink_state_root_rejected(self): + self.root.mkdir() + alias = self.base / "alias" + alias.symlink_to(self.root, target_is_directory=True) + with self.assertRaises(t.TestbedError): + t.state_dir(str(alias)) + + def test_state_inside_repo_rejected(self): + with self.assertRaises(t.TestbedError): + t.state_dir(str(t.REPO / "local")) + + def test_plan_is_pure(self): + with patch.object(t, "local_command", side_effect=AssertionError("execution forbidden")): + result = t.plan(self.cfg, self.root, self.iso) + self.assertEqual(result["status"], "PLANNED_NOT_EXECUTED") + self.assertFalse(self.root.exists()) + + def test_failed_subprocess_retains_stderr_and_stdout(self): + failure = subprocess.CalledProcessError(7, ["virsh", "domstate"], + output="partial output", + stderr="permission denied") + with patch.object(t.subprocess, "run", side_effect=failure): + with self.assertRaises(t.TestbedError) as caught: + t.local_command(["virsh", "domstate"]) + self.assertIn("partial output", str(caught.exception)) + self.assertIn("permission denied", str(caught.exception)) + self.assertIn("exit status 7", str(caught.exception)) + + def test_timeout_preserves_partial_diagnostics(self): + failure = subprocess.TimeoutExpired(["qemu-img", "convert"], 3, + output=b"partial\\xff", + stderr=b"disk full") + with patch.object(t.subprocess, "run", side_effect=failure): + with self.assertRaises(t.TestbedError) as caught: + t.local_command(["qemu-img", "convert"]) + self.assertIn("timed out", str(caught.exception)) + self.assertIn("disk full", str(caught.exception)) + + def test_prepare_requires_operator_approval(self): + with self.assertRaises(t.TestbedError): + t.prepare(self.cfg, self.root, self.iso, None) + self.assertFalse(self.root.exists()) + + def test_prepare_never_defines_or_boots_vm(self): + def fake_qemu(args, timeout=30): + self.assertEqual(args[:4], ["qemu-img", "create", "-f", "qcow2"]) + Path(args[4]).write_bytes(b"simulated image") + return "" + with patch.object(t, "need"), patch.object(t, "local_command", side_effect=fake_qemu): + receipt = t.prepare(self.cfg, self.root, self.iso, t.PREPARE_TOKEN) + self.assertEqual(receipt["status"], "PREPARED_NOT_DEFINED_OR_BOOTED") + self.assertTrue((self.root / "domain.xml").is_file()) + self.assertEqual(os.stat(self.root).st_mode & 0o077, 0) + with self.assertRaises(t.TestbedError): + t.prepare(self.cfg, self.root, self.iso, t.PREPARE_TOKEN) + + def test_snapshot_requires_approval(self): + with self.assertRaises(t.TestbedError): + t.snapshot(self.cfg, self.root, self.iso, "baseline", "a" * 40, None) + + def test_snapshot_refuses_live_vm(self): + self.root.mkdir(mode=0o700) + (self.root / "disk.qcow2").write_bytes(b"fixture") + with patch.object(t, "need"), patch.object(t, "audit_defined"), \ + patch.object(t, "local_command", return_value="running"): + with self.assertRaises(t.TestbedError): + t.snapshot(self.cfg, self.root, self.iso, "baseline", "a" * 40, + t.SNAPSHOT_TOKEN) + self.assertFalse((self.root / "snapshots").exists()) + + def test_snapshot_rejects_bad_sha(self): + with self.assertRaises(t.TestbedError): + t.snapshot(self.cfg, self.root, self.iso, "baseline", "not-a-sha", + t.SNAPSHOT_TOKEN) + + def test_audit_checks_both_live_and_persistent(self): + safe = t.render_xml(self.cfg, self.root, self.iso) + calls = [] + def fake_command(args, timeout=30): + calls.append(args) + return safe + with patch.object(t, "need"), patch.object(t, "local_command", + side_effect=fake_command): + t.audit_defined(self.cfg, self.root, self.iso) + self.assertEqual(len(calls), 2) + self.assertIn("--inactive", calls[0]) + self.assertNotIn("--inactive", calls[1]) + + +if __name__ == "__main__": + unittest.main()