diff options
| -rw-r--r-- | .chezmoiignore | 1 | ||||
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | docs/canonical-vm.md | 166 | ||||
| -rw-r--r-- | just-lib.sh | 6 | ||||
| -rw-r--r-- | justfile | 61 | ||||
| -rw-r--r-- | nix/flake.nix | 9 | ||||
| -rw-r--r-- | pyrightconfig.json | 1 | ||||
| -rw-r--r-- | scripts/canonical-vm-guest.sh | 101 | ||||
| -rw-r--r-- | scripts/canonical-vm-install.sh | 50 | ||||
| -rw-r--r-- | scripts/canonical-vm-keyring.sh | 28 | ||||
| -rw-r--r-- | scripts/canonical.py | 36 | ||||
| -rw-r--r-- | scripts/canonical_vm.py | 802 | ||||
| -rw-r--r-- | tests/test_canonical.py | 40 | ||||
| -rw-r--r-- | tests/test_canonical_vm.py | 230 |
14 files changed, 1521 insertions, 11 deletions
diff --git a/.chezmoiignore b/.chezmoiignore index 4c4ad4f..28568bb 100644 --- a/.chezmoiignore +++ b/.chezmoiignore @@ -18,6 +18,7 @@ just-lib.sh selene.toml selene-globals.yml scripts/ +tmp/ tests/ docs/ canonical/ @@ -1,6 +1,7 @@ .direnv/ !home/.lldbinit .worktrees/ +tmp/ .ruff_cache/ node_modules/ *.swp diff --git a/docs/canonical-vm.md b/docs/canonical-vm.md new file mode 100644 index 0000000..502eae7 --- /dev/null +++ b/docs/canonical-vm.md @@ -0,0 +1,166 @@ +# Canonical Desktop Test VM + +This lab uses Ubuntu 26.04.1 Desktop, GNOME/GDM, and the existing `canonical` +role. It does not enroll in Landscape or authd. Do not put the corporate +autoinstall file or company credentials in this lab. + +## Host Access + +Run outside aibox as your normal user with access to the local LXD socket. +LXD access is equivalent to root access. Do not mount its socket into aibox. +Temporary group access does not require permanent group membership: + +```sh +sudo -u "$USER" -g lxd -- "$HOME/.nix-profile/bin/codex" resume \ + --sandbox danger-full-access --ask-for-approval never +``` + +On Arch, LXD's VM support needs `cdrtools` and its QEMU/OVMF dependencies. +Install missing host packages with pacman before starting. Image and screenshot +tools are in the repo's Nix development shell: + +```sh +nix develop ./nix +just canonical-vm-run +``` + +If the main Ubuntu download server is slow, `canonical-vm-create` accepts an +HTTPS mirror directory as its argument. It resumes a partial ISO download and +still checks the ISO against the signed manifest from `releases.ubuntu.com`. + +The lab uses `canonical-lab`, 4 CPUs, 8 GiB RAM, an 80 GiB disk, and the existing +`default` storage pool and `lxdbr0` bridge. It does not change shared profiles, +firewall rules, or other instances. Guests can reach the LAN and VPN through +the existing bridge. No host home directory or credential agent is shared. +Allow several hours for a cold run. The desktop runtimes are large, and the +shared Nix profile can build packages from source. + +## Test Credentials + +The workflow creates test SSH and disk/login credentials under +`tmp/canonical-vm/secrets/`. The lab directory is private and ignored by both +Git and chezmoi. It also holds the verified ISO, private installer, and reports. +Keep it for retries. Do not publish it or add it to Git. + +The VM has encrypted LVM, but its unlock key is in the unencrypted initramfs. +It has passwordless sudo and GDM automatic login. These settings are for this +disposable VM only. They do not provide a secure laptop configuration. +The lab also locks `clock-show-weekday` to false to test policy handling. +The lab keyring service reads the generated login password from a private file +at startup. Bootstrap saves the initial empty keyring before creating the test +keyring. It does not change the corporate role's keyring or PAM configuration. + +SSH uses the generated key only, with agent forwarding disabled. Its host key +is read through the trusted local LXD agent and then pinned. A different host +key stops the workflow. The guest creates a dummy GPG signing key. +Its private export stays in the local secrets directory for snapshot retries. + +## Operations + +`canonical-vm-run` downloads and verifies the Ubuntu ISO, adds the lab installer +data, installs Ubuntu, takes the `clean` snapshot, installs upstream multi-user +Nix, takes `nix-ready`, and runs `canonical-setup` inside the guest. It then +reboots, runs the guest checks, stops the VM, and takes `working`. + +The source copy includes current tracked changes and untracked, nonignored +files. It excludes host Git state, lab data, caches, and worktrees. The guest +gets a separate Git repository. No setup recipe runs on the host. + +Individual steps are available when a test fails: + +```sh +just canonical-vm-status +just canonical-vm-console +just canonical-vm-screenshot +just canonical-vm-logs +just canonical-vm-stop +just canonical-vm-restore nix-ready +just canonical-vm-start +just canonical-vm-sync-source +just canonical-vm-deploy +just canonical-vm-test +``` + +Restore discards changes in the marked test VM. Snapshots require a stopped VM. +Existing snapshots are not overwritten. Commands reject an instance whose +ownership marker does not match the local lab state. +`canonical-vm-sync-source` discards edits in the guest source checkout. It does +not change the guest's keys or deployed home files. Run it before deployment +when testing a source fix after restoring `nix-ready`. + +`canonical-vm-delete` removes only the marked VM and its snapshots. It leaves +the private local lab files for inspection. Remove `tmp/canonical-vm/` yourself +after you no longer need them. Never reuse its credentials on another system. +Remove the cached `installer.iso` before creating another VM when you change +the installer scripts. Keep the original Ubuntu ISO to avoid another download. + +## Test Boundary + +Guest checks use the real GNOME user session. `canonical-lab-check` requires +Ubuntu, the `canonical` role, a VM, and a root-owned lab marker. It skips company +registration only. The normal `canonical-check` still checks Landscape. + +The VM cannot verify company policy, Google/authd login, normal GDM password +login, or laptop hardware. Camera, dock, suspend, and real-account notification +tests still need the laptop. Autologin and the test keyring service do not test +PAM integration. A successful command alone does not prove a visible GUI action. + +## Test Results (2026-09-14) + +The installed VM boots with encrypted LVM and automatic unlock. The full Nix +profile, Snaps, and Flatpaks installed. All configured GNOME extensions were +active after settings deployment. The policy-lock, GPG signing, rootless Podman, +aibox, and secret-storage checks passed. The stored secret survived a reboot. A screenshot +confirmed the desktop panel and test notification. The source checks and all +unit tests passed. + +Keybase Snap 6.5.1 revision 70 corrupted the GNOME settings database when its +GUI started. Its `setup-env.sh` sets `XDG_RUNTIME_DIR` to its `.config` +directory, whose `dconf/user` links to the desktop database. A controlled +launch changed the header from `GVariant` to `G\0ariant`. On later boots, +dconf discarded the database and the desktop extensions disappeared. +Keybase is removed from the VM and excluded from the corporate laptop profile. +After removal, the deployment and final reboot checks passed. All configured +extensions stayed active without reapplying settings, and the stored test +secret survived. The final screenshot confirmed that the panel remained visible. + +These limits remain: + +- Mattermost remains a Snap. Revision 850 lacks the keyring plug. The local + Mattermost-only AppArmor rule permits Secret Service access, and the app + reports `encryption available`. AppArmor remains enforced. The path watcher + restored the rule after a simulated profile replacement. Both subsequent + reboot checks passed with encryption available. Real-account login was not + tested. See [Mattermost keyring](canonical-laptop.md#mattermost-keyring). +- Flatpak reported end-of-life KDE runtimes for Nheko (6.8) and NormCap (6.9). + The installed stable releases still use these runtimes. + +The crates.io HTTP 403 failure is fixed in the Nix package definition. Tuicr's +crate downloads use `static.crates.io` and retain the Cargo.lock checksums. +All 377 crate downloads were forced in the Ubuntu VM, without reusing their +cached outputs, and passed hash verification. The Nix profile build also passed. +No manual crate prefetch is needed. Run `just nix-crate-check` to repeat a +forced download of one locked crate. The full download test log is +`tmp/canonical-vm/reports/tuicr-forced-downloads.log`. + +The full guest `just update` passed after GitHub's API rate limit cleared. +This included the package updates, release check, Nix flake update and +activation, Neovim update, and GNOME extension update. No new APT packages +were installed. The download checks above cover the crate-cache failure; they +do not repeat the OS installation or company provisioning. +Use login Zsh for CLI tests, as Ghostty does; Bash does not load the Home +Manager session variables used by the parser compiler. + +The stopped `working` snapshot restored the guest disk state: a test file added +after the snapshot was absent after restore. An early reboot test caused a +GNOME shutdown timeout and activated its extension failure protection. Desktop +readiness now waits for GNOME's extension startup check to finish. It does not +disable that protection. The final reboot check also verifies extensions +without applying settings again. The `working` snapshot was replaced after the +update and reboot tests. It excludes Keybase and includes the Mattermost +keyring workaround and the crate download fix. The VM is stopped. Logs and screenshots are under +`tmp/canonical-vm/reports/`. + +References: [Ubuntu images](https://releases.ubuntu.com/26.04/), +[autoinstall](https://canonical-subiquity.readthedocs-hosted.com/en/latest/reference/autoinstall-reference.html), +and [LXD VMs](https://canonical.com/lxd/docs/latest/howto/instances_create/). diff --git a/just-lib.sh b/just-lib.sh index 5caa79f..dadf427 100644 --- a/just-lib.sh +++ b/just-lib.sh @@ -15,18 +15,18 @@ _find_shells() { \( -name '*.sh' \ -o -path './dot_local/bin/executable_*' \ -o -path './dot_config/sway/executable_*' \) \ - -not -path './.git/*' -not -path './.worktrees/*' + -not -path './.git/*' -not -path './.worktrees/*' -not -path './tmp/*' } _find_by_ext() { find . -type f -name "*.$1" \ - -not -path './.git/*' -not -path './.worktrees/*' + -not -path './.git/*' -not -path './.worktrees/*' -not -path './tmp/*' } _find_zsh() { find . -type f \ \( -name 'dot_zshrc' -o -name 'dot_zshenv' -o -name 'dot_zprofile' \) \ - -not -path './.git/*' -not -path './.worktrees/*' + -not -path './.git/*' -not -path './.worktrees/*' -not -path './tmp/*' } _is_zsh() { @@ -28,6 +28,65 @@ nix-switch: canonical-check: @python3 scripts/canonical.py check +# Disposable Ubuntu desktop VM. Never deploys onto the host. +canonical-vm-create mirror="https://releases.ubuntu.com/26.04/": + @python3 scripts/canonical_vm.py create --mirror {{ quote(mirror) }} + +canonical-vm-status: + @python3 scripts/canonical_vm.py status + +canonical-vm-console: + @python3 scripts/canonical_vm.py console + +canonical-vm-start: + @python3 scripts/canonical_vm.py start + +canonical-vm-stop: + @python3 scripts/canonical_vm.py stop + +canonical-vm-finish-install: + @python3 scripts/canonical_vm.py finish-install + +canonical-vm-snapshot name: + @python3 scripts/canonical_vm.py snapshot {{ quote(name) }} + +canonical-vm-restore name: + @python3 scripts/canonical_vm.py restore {{ quote(name) }} + +canonical-vm-delete: + @python3 scripts/canonical_vm.py delete + +canonical-lab-check: + @python3 scripts/canonical.py lab-check + +canonical-vm-bootstrap: + @python3 scripts/canonical_vm.py bootstrap + +canonical-vm-test: + @python3 scripts/canonical_vm.py test + +canonical-vm-logs: + @python3 scripts/canonical_vm.py logs + +canonical-vm-screenshot: + @python3 scripts/canonical_vm.py screenshot + +canonical-vm-deploy: + @python3 scripts/canonical_vm.py deploy + +# Replace the marked guest's source checkout with the current host source. +canonical-vm-sync-source: + @python3 scripts/canonical_vm.py sync-source + +# Create a new lab, test deployment, and leave it stopped with a working snapshot. +canonical-vm-run: canonical-vm-create _canonical-vm-wait canonical-vm-finish-install canonical-vm-start canonical-vm-bootstrap _canonical-vm-reboot canonical-vm-test canonical-vm-stop (canonical-vm-snapshot "working") + +_canonical-vm-wait: + @python3 scripts/canonical_vm.py wait-install + +_canonical-vm-reboot: + @python3 scripts/canonical_vm.py reboot + # Install corporate packages, profile, dotfiles, and desktop settings. canonical-setup: _require-canonical _install-hooks (pkg-apply "base") nix-switch apply canonical-system canonical-extensions canonical-desktop @@ -1390,7 +1449,7 @@ _desktop-update: set -eu source just-lib.sh [ "$(_machine_role)" = canonical ] || exit 0 - gext update --install $(sed '/^#/d; /^$/d' meta/canonical/extensions.txt) + gext --filesystem update --install $(sed '/^#/d; /^$/d' meta/canonical/extensions.txt) _canonical-finish: #!/usr/bin/env bash diff --git a/nix/flake.nix b/nix/flake.nix index 33dbe42..a0f392a 100644 --- a/nix/flake.nix +++ b/nix/flake.nix @@ -285,6 +285,7 @@ python3 just nixfmt + nodejs prettier ruff selene @@ -292,6 +293,14 @@ shfmt stylua taplo + xorriso + spice-gtk + aria2 + curl + gnupg + openssh + openssl + imagemagick ]; }; diff --git a/pyrightconfig.json b/pyrightconfig.json index 7b59b39..54edcb6 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -2,6 +2,7 @@ "typeCheckingMode": "standard", "include": ["**/*.py"], "exclude": [ + "tmp/**", "**/.git/**", "**/.venv/**", "**/node_modules/**", diff --git a/scripts/canonical-vm-guest.sh b/scripts/canonical-vm-guest.sh new file mode 100644 index 0000000..29caa7d --- /dev/null +++ b/scripts/canonical-vm-guest.sh @@ -0,0 +1,101 @@ +#!/bin/bash +set -euo pipefail +test -f /etc/canonical-lab +test "$(id -un)" = canonical-test +# shellcheck source=/dev/null +test "$( + . /etc/os-release + printf '%s' "$ID" +)" = ubuntu +systemd-detect-virt --vm --quiet +cd "$HOME/dotfiles" + +check_extensions() { + test "$(/usr/bin/gsettings get org.gnome.shell disable-user-extensions)" = false + for extension in corporate-panel@dotfiles $(sed '/^#/d; /^$/d' meta/canonical/extensions.txt); do + gnome-extensions info "$extension" + gnome-extensions list --enabled | grep -Fx "$extension" + gnome-extensions info "$extension" | grep -Eq 'State: (ACTIVE|ENABLED)$' + done +} + +check_mattermost_keyring() { + systemctl is-active dotfiles-mattermost-keyring.path + snap run --shell mattermost-desktop <<'EOF' +dbus-send --session --print-reply --dest=org.freedesktop.secrets /org/freedesktop/secrets org.freedesktop.DBus.Peer.Ping +EOF + grep 'Secure storage initialized' "$HOME/snap/mattermost-desktop/current/.config/Mattermost/logs/main.log" | + tail -n 1 | grep -F 'encryption available' +} + +case "${1:?stage required}" in + session) + check_extensions + test "$(secret-tool lookup application canonical-lab)" = lab-value + check_mattermost_keyring + ;; + nix) + if ! test -x /nix/var/nix/profiles/default/bin/nix; then + curl --fail --location https://nixos.org/nix/install -o /tmp/install-nix + sh /tmp/install-nix --daemon --yes --no-channel-add + fi + ;; + identity) + install -d -m 700 "$HOME/.gnupg" "$HOME/.config/chezmoi" + if ! gpg --list-secret-keys canonical-test@example.invalid >/dev/null 2>&1; then + gpg --batch --pinentry-mode loopback --passphrase '' --quick-generate-key \ + 'Canonical Test <canonical-test@example.invalid>' ed25519 sign 1y + fi + fingerprint=$(gpg --with-colons --list-secret-keys canonical-test@example.invalid | awk -F: '$1=="fpr" {print $10; exit}') + cat >"$HOME/.config/chezmoi/chezmoi.toml" <<EOF +sourceDir = "$HOME/dotfiles" +[data] +machineRole = "canonical" +workName = "Canonical Test" +workEmail = "canonical-test@example.invalid" +workSigningKey = "$fingerprint" +EOF + git init -q + git -c user.name='Canonical Test' -c user.email=canonical-test@example.invalid \ + -c commit.gpgsign=false add . + git -c core.hooksPath=/dev/null -c user.name='Canonical Test' -c user.email=canonical-test@example.invalid \ + -c commit.gpgsign=false commit --allow-empty -qm 'Import lab source' + ;; + setup) + export NIX_CONFIG='experimental-features = nix-command flakes' + export PATH="/nix/var/nix/profiles/default/bin:$HOME/.nix-profile/bin:$PATH" + nix develop ./nix --command just canonical-setup + ;; + settings) + export PATH="$HOME/.nix-profile/bin:/nix/var/nix/profiles/default/bin:$PATH" + just canonical-desktop-restore + just canonical-desktop + test "$(gsettings get org.gnome.desktop.interface clock-show-weekday)" = false + ;; + check) + export PATH="$HOME/.nix-profile/bin:/nix/var/nix/profiles/default/bin:$PATH" + test -n "${WAYLAND_DISPLAY:-}" + test "${XDG_CURRENT_DESKTOP#*GNOME}" != "$XDG_CURRENT_DESKTOP" + just canonical-desktop + test "$(gsettings writable org.gnome.desktop.interface clock-show-weekday)" = false + test "$(gsettings get org.gnome.desktop.interface clock-show-weekday)" = false + just canonical-lab-check + just apply + # shellcheck disable=SC2016 + "$HOME/.nix-profile/bin/zsh" -lc 'test -x "$NVIM_TREESITTER_CC"' + lsblk -sno TYPE "$(findmnt -no SOURCE /)" | grep -Fx crypt + gpg --batch --local-user canonical-test@example.invalid --output /tmp/lab-signature --detach-sign --yes README.md + gpg --verify /tmp/lab-signature README.md + podman run --rm docker.io/library/alpine:3.22 true + # shellcheck disable=SC2016 + aibox -- sh -c 'test "$AIBOX" = 1 && test ! -S /var/lib/lxd/unix.socket' + check_extensions + busctl --user get-property org.freedesktop.secrets /org/freedesktop/secrets/collection/login \ + org.freedesktop.Secret.Collection Locked | grep -Fx 'b false' + printf 'lab-value' | timeout 30 secret-tool store --label='Canonical lab test' application canonical-lab + test "$(secret-tool lookup application canonical-lab)" = lab-value + check_mattermost_keyring + notify-send 'Canonical lab' 'Notification test' + ;; + *) exit 2 ;; +esac diff --git a/scripts/canonical-vm-install.sh b/scripts/canonical-vm-install.sh new file mode 100644 index 0000000..177d06e --- /dev/null +++ b/scripts/canonical-vm-install.sh @@ -0,0 +1,50 @@ +#!/bin/bash +set -euo pipefail +systemd-detect-virt --vm --quiet +test -f /cdrom/lab/owner + +mount_config() { + mkdir -p "$1" + modprobe 9pnet_virtio + mount -t 9p config "$1" -o access=0,trans=virtio || + mount -t virtiofs config "$1" +} + +if [ "${1:-}" = live ]; then + mount_config /mnt/lxd-agent + (cd /mnt/lxd-agent && ./install.sh) + umount /mnt/lxd-agent + systemctl start lxd-agent + exit 0 +fi + +test -d /target/etc +install -m 600 /cdrom/lab/owner /target/etc/canonical-lab +install -m 600 /cdrom/lab/disk-password /target/root/lab-unlock.key +# The disposable guest stores its unlock key in the unencrypted initramfs. +awk 'NF && $1 !~ /^#/ {$3="/root/lab-unlock.key"} {print}' \ + /target/etc/crypttab >/target/etc/crypttab.lab +mv /target/etc/crypttab.lab /target/etc/crypttab +mkdir -p /target/etc/dracut.conf.d +printf 'install_items+=" /root/lab-unlock.key "\n' >/target/etc/dracut.conf.d/99-canonical-lab.conf +printf 'canonical-test ALL=(ALL:ALL) NOPASSWD:ALL\n' >/target/etc/sudoers.d/canonical-lab +chmod 440 /target/etc/sudoers.d/canonical-lab +printf '[daemon]\nAutomaticLoginEnable=True\nAutomaticLogin=canonical-test\n' >/target/etc/gdm3/custom.conf +printf 'PermitRootLogin no\nPasswordAuthentication no\nAllowAgentForwarding no\n' >/target/etc/ssh/sshd_config.d/00-canonical-lab.conf +printf 'APT::Get::Assume-Yes "true";\n' >/target/etc/apt/apt.conf.d/99canonical-lab +mkdir -p /target/etc/dconf/profile /target/etc/dconf/db/canonical-lab.d/locks +if ! test -f /target/etc/dconf/profile/user; then + printf 'user-db:user\n' >/target/etc/dconf/profile/user +fi +printf 'system-db:canonical-lab\n' >>/target/etc/dconf/profile/user +printf '[org/gnome/desktop/interface]\nclock-show-weekday=false\n' >/target/etc/dconf/db/canonical-lab.d/00-settings +printf '/org/gnome/desktop/interface/clock-show-weekday\n' >/target/etc/dconf/db/canonical-lab.d/locks/test +curtin in-target --target=/target -- dconf update + +mount_config /target/mnt/lxd-agent +curtin in-target --target=/target -- bash -c 'cd /mnt/lxd-agent && ./install.sh' +umount /target/mnt/lxd-agent +curtin in-target --target=/target -- update-initramfs -u -k all +# shellcheck disable=SC2016 +curtin in-target --target=/target -- bash -ec \ + 'for image in /boot/initrd.img-*; do lsinitrd "$image" | grep -F root/lab-unlock.key; done' diff --git a/scripts/canonical-vm-keyring.sh b/scripts/canonical-vm-keyring.sh new file mode 100644 index 0000000..8af901a --- /dev/null +++ b/scripts/canonical-vm-keyring.sh @@ -0,0 +1,28 @@ +#!/bin/bash +set -euo pipefail +test "$(id -u)" = 0 +test -f /etc/canonical-lab +systemd-detect-virt --vm --quiet +home=/home/canonical-test +key=$home/.local/state/canonical-lab/keyring-password +test ! -f "$key" || exit 0 +systemctl stop gdm +loginctl terminate-user canonical-test +umask 077 +install -d -o canonical-test -g canonical-test "$home/.local/state/canonical-lab" +if test -d "$home/.local/share/keyrings"; then + test ! -e "$home/.local/share/keyrings-before-lab-unlock" + mv "$home/.local/share/keyrings" "$home/.local/share/keyrings-before-lab-unlock" +fi +cat >"$key" +test -s "$key" +chown canonical-test:canonical-test "$key" +dropin=$home/.config/systemd/user/gnome-keyring-daemon.service.d +install -d -o canonical-test -g canonical-test "$dropin" +cat >"$dropin/lab.conf" <<'EOF' +[Service] +ExecStart= +ExecStart=/usr/bin/gnome-keyring-daemon --foreground --components=pkcs11,secrets --control-directory=%t/keyring --unlock +StandardInput=file:%h/.local/state/canonical-lab/keyring-password +EOF +chown canonical-test:canonical-test "$dropin/lab.conf" diff --git a/scripts/canonical.py b/scripts/canonical.py index 2ff6310..db7f773 100644 --- a/scripts/canonical.py +++ b/scripts/canonical.py @@ -81,14 +81,28 @@ def install() -> None: subprocess.run(command, check=True) -def check() -> None: +def require_lab() -> None: + marker = Path("/etc/canonical-lab") + if not marker.is_file() or marker.stat().st_uid != 0: + raise SystemExit("This command requires a root-owned lab marker.") + if subprocess.run( + ["systemd-detect-virt", "--vm", "--quiet"], check=False + ).returncode: + raise SystemExit("This command requires a virtual machine.") + + +def check(lab: bool = False) -> None: + if lab: + require_lab() + print("UNTESTED: company provisioning, authd, and Landscape registration.") commands = [ ["lsb_release", "-ds"], *[ ["systemctl", "is-active", unit] - for unit in ["display-manager", "snapd", "apparmor", "nix-daemon"] + for unit in ["display-manager", "snapd", "apparmor"] ], - ["landscape-config", "--actively-registered"], + ["nix", "store", "ping", "--store", "daemon"], + *([] if lab else [["landscape-config", "--actively-registered"]]), ["snap", "connections", "thunderbird"], ["snap", "list", *packages("snap")], ["gnome-extensions", "list", "--enabled"], @@ -113,14 +127,22 @@ def check() -> None: def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( - "action", choices=["install", "update", "flatpak-update", "extensions", "check"] + "action", + choices=[ + "install", + "update", + "flatpak-update", + "extensions", + "check", + "lab-check", + ], ) args = parser.parse_args() require_canonical() if args.action == "install": install() - elif args.action == "check": - check() + elif args.action in {"check", "lab-check"}: + check(lab=args.action == "lab-check") else: commands = update_commands() if args.action == "flatpak-update": @@ -128,7 +150,7 @@ def main() -> None: ["flatpak", "update", "--user", "--assumeyes", *packages("flatpak")] ] elif args.action == "extensions": - commands = [["gext", "install", *packages("extensions")]] + commands = [["gext", "--filesystem", "install", *packages("extensions")]] for command in commands: subprocess.run(command, check=True) diff --git a/scripts/canonical_vm.py b/scripts/canonical_vm.py new file mode 100644 index 0000000..b8af4df --- /dev/null +++ b/scripts/canonical_vm.py @@ -0,0 +1,802 @@ +"""Create and operate the disposable Canonical desktop test VM.""" + +import argparse +import hashlib +import json +import os +import re +import secrets +import select +import shlex +import shutil +import subprocess +import tarfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +STATE = ROOT / "tmp/canonical-vm" +NAME = "canonical-lab" +MARKER = "user.dotfiles-lab" +ISO = "ubuntu-26.04.1-desktop-amd64.iso" +RELEASE = "https://releases.ubuntu.com/26.04/" +SIGNER = "843938DF228D22F7B3742BC0D94AA3F0EFE21092" + + +def run(*args: str, **kwargs): + return subprocess.run(args, check=True, **kwargs) + + +def output(*args: str) -> str: + return subprocess.check_output(args, text=True).strip() + + +def lxc(*args: str, **kwargs): + return run("lxc", "--force-local", "--project", "default", *args, **kwargs) + + +def query(path: str): + return json.loads( + output("lxc", "--force-local", "query", path + "?project=default") + ) + + +def require_owned(instance: dict, owner: str) -> None: + if ( + instance.get("type") != "virtual-machine" + or instance.get("config", {}).get(MARKER) != owner + ): + raise RuntimeError("Refusing to change an unmarked VM.") + + +def owner() -> str: + return (STATE / "owner").read_text().strip() + + +def checked_instance(): + instance = query("/1.0/instances/" + NAME) + require_owned(instance, owner()) + return instance + + +def private_file(path: Path, data: str) -> None: + with path.open("x", encoding="utf-8") as stream: + stream.write(data) + path.chmod(0o600) + + +def prepare() -> None: + if os.environ.get("AIBOX"): + raise RuntimeError("Run the VM workflow outside aibox.") + run("git", "check-ignore", "-q", "tmp/canonical-vm/secrets/probe", cwd=ROOT) + if "tmp/" not in (ROOT / ".chezmoiignore").read_text().splitlines(): + raise RuntimeError("The lab directory must be ignored by chezmoi.") + if STATE.resolve() != STATE or (STATE / "secrets").is_symlink(): + raise RuntimeError("The lab directory must not contain redirected paths.") + os.umask(0o077) + STATE.mkdir(parents=True, exist_ok=True, mode=0o700) + STATE.chmod(0o700) + if not (STATE / "owner").exists(): + private_file(STATE / "owner", secrets.token_hex(16)) + (STATE / "secrets").mkdir(exist_ok=True, mode=0o700) + for name in ["login-password", "disk-password"]: + path = STATE / "secrets" / name + if not path.exists(): + private_file(path, secrets.token_urlsafe(32)) + key = STATE / "secrets/ssh" + if not key.exists(): + run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", NAME, "-f", str(key)) + + +def download(name: str, mirror: str = RELEASE) -> None: + target = STATE / name + if not target.exists(): + if name == ISO: + run( + "aria2c", + "--continue=true", + "--max-connection-per-server=8", + "--split=8", + "--auto-file-renaming=false", + "--summary-interval=60", + "--console-log-level=warn", + "--dir=" + str(STATE), + "--out=" + name + ".part", + mirror.rstrip("/") + "/" + name, + ) + Path(str(target) + ".part").rename(target) + return + run( + "curl", + "--fail", + "--location", + "--retry", + "3", + "--output", + str(target) + ".part", + RELEASE + name, + ) + Path(str(target) + ".part").rename(target) + + +def verify_iso(mirror: str = RELEASE) -> None: + for name in ["SHA256SUMS", "SHA256SUMS.gpg"]: + download(name) + keyring = STATE / "ubuntu-keyring" + keyring.mkdir(exist_ok=True, mode=0o700) + run( + "gpg", + "--homedir", + str(keyring), + "--batch", + "--keyserver", + "hkps://keyserver.ubuntu.com", + "--recv-keys", + SIGNER, + ) + run( + "gpg", + "--homedir", + str(keyring), + "--batch", + "--verify", + str(STATE / "SHA256SUMS.gpg"), + str(STATE / "SHA256SUMS"), + ) + expected = next( + line.split()[0] + for line in (STATE / "SHA256SUMS").read_text().splitlines() + if line.split()[-1].lstrip("*") == ISO + ) + download(ISO, mirror) + with (STATE / ISO).open("rb") as stream: + actual = hashlib.file_digest(stream, "sha256").hexdigest() + if actual != expected: + raise RuntimeError("Ubuntu ISO checksum does not match its signed manifest.") + + +def autoinstall(password_hash: str, public_key: str, disk_password: str) -> dict: + return { + "autoinstall": { + "version": 1, + "refresh-installer": {"update": False}, + "early-commands": [ + ["systemd-detect-virt", "--vm", "--quiet"], + ["bash", "/cdrom/lab/install.sh", "live"], + ], + "locale": "en_US.UTF-8", + "keyboard": {"layout": "us"}, + "timezone": "Europe/Lisbon", + "identity": { + "hostname": NAME, + "username": "canonical-test", + "realname": "Canonical Test", + "password": password_hash, + }, + "storage": { + "layout": { + "name": "lvm", + "match": {"path": "/dev/sda"}, + "password": disk_password, + "sizing-policy": "all", + } + }, + "ssh": { + "install-server": True, + "allow-pw": False, + "authorized-keys": [public_key], + }, + "packages": [ + "openssh-server", + "git", + "curl", + "gnome-keyring", + "libsecret-tools", + "gnupg", + ], + "late-commands": [["bash", "/cdrom/lab/install.sh"]], + "shutdown": "poweroff", + } + } + + +def boot_config(text: str) -> str: + patched, count = re.subn( + r"(linux\s+/casper/vmlinuz[^\n]*?)\s+---", r"\1 autoinstall ---", text + ) + if not count: + raise RuntimeError("Unknown Ubuntu installer boot configuration.") + return re.sub(r"set timeout=\d+", "set timeout=1", patched) + + +def installer(mirror: str = RELEASE) -> None: + prepare() + verify_iso(mirror) + media = STATE / "media" + media.mkdir(exist_ok=True, mode=0o700) + password_hash = ( + subprocess.check_output( + ["openssl", "passwd", "-6", "-stdin"], + input=(STATE / "secrets/login-password").read_bytes(), + ) + .decode() + .strip() + ) + config = autoinstall( + password_hash, + (STATE / "secrets/ssh.pub").read_text().strip(), + (STATE / "secrets/disk-password").read_text(), + ) + (media / "autoinstall.yaml").write_text(json.dumps(config, indent=2)) + run( + "xorriso", + "-osirrox", + "on", + "-indev", + str(STATE / ISO), + "-extract", + "/boot/grub/grub.cfg", + str(media / "grub.cfg"), + ) + (media / "grub.cfg").chmod(0o600) + (media / "grub.cfg").write_text(boot_config((media / "grub.cfg").read_text())) + target = STATE / "installer.iso" + if target.exists(): + raise RuntimeError( + "Installer already exists; keep it for retries or remove it explicitly." + ) + run( + "xorriso", + "-indev", + str(STATE / ISO), + "-outdev", + str(target) + ".part", + "-boot_image", + "any", + "replay", + "-map", + str(media / "grub.cfg"), + "/boot/grub/grub.cfg", + "-map", + str(media / "autoinstall.yaml"), + "/autoinstall.yaml", + "-map", + str(ROOT / "scripts/canonical-vm-install.sh"), + "/lab/install.sh", + "-map", + str(STATE / "secrets/disk-password"), + "/lab/disk-password", + "-map", + str(STATE / "owner"), + "/lab/owner", + ) + Path(str(target) + ".part").rename(target) + + +def create(mirror: str = RELEASE) -> None: + prepare() + instances = query("/1.0/instances") + if any(path.split("?")[0].endswith("/" + NAME) for path in instances): + checked_instance() + raise RuntimeError("Lab VM already exists; use start, status, or restore.") + if not (STATE / "installer.iso").exists(): + installer(mirror) + lxc( + "init", + NAME, + "--empty", + "--vm", + "--no-profiles", + "-s", + "default", + "-c", + MARKER + "=" + owner(), + "-c", + "limits.cpu=4", + "-c", + "limits.memory=8GiB", + "-d", + "root,size=80GiB", + ) + lxc("config", "device", "add", NAME, "eth0", "nic", "network=lxdbr0", "name=eth0") + lxc( + "config", + "device", + "add", + NAME, + "installer", + "disk", + "source=" + str(STATE / "installer.iso"), + "boot.priority=10", + ) + lxc("start", NAME) + print("Installer started. Use canonical-vm-status or canonical-vm-console.") + + +def source_allowed(path: str) -> bool: + return not any( + part in {"tmp", ".git", ".direnv", ".worktrees", "node_modules", "__pycache__"} + for part in Path(path).parts + ) + + +def source_archive() -> Path: + files = ( + subprocess.check_output( + ["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z"], + cwd=ROOT, + ) + .decode() + .split("\0") + ) + archive = STATE / "source.tar" + with tarfile.open(archive, "w") as stream: + for name in sorted(set(files)): + if name and source_allowed(name) and (ROOT / name).exists(): + stream.add(ROOT / name, arcname=name, recursive=False) + return archive + + +def guest(*args: str, **kwargs): + checked_instance() + return lxc("exec", NAME, "--", *args, **kwargs) + + +def wait_agent() -> None: + checked_instance() + deadline = time.monotonic() + 600 + while time.monotonic() < deadline: + result = subprocess.run( + [ + "lxc", + "--force-local", + "--project", + "default", + "exec", + NAME, + "--", + "true", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + if result.returncode == 0: + return + time.sleep(5) + raise RuntimeError("Guest agent did not become ready within ten minutes.") + + +def snapshot(name: str) -> None: + instance = checked_instance() + if instance["status"] != "Stopped": + raise RuntimeError("Stop the lab VM before taking a snapshot.") + lxc("snapshot", NAME, name) + + +def stop() -> None: + if checked_instance()["status"] == "Stopped": + return + guest("systemctl", "poweroff", "--no-block") + deadline = time.monotonic() + 180 + while time.monotonic() < deadline: + if checked_instance()["status"] == "Stopped": + return + time.sleep(5) + raise RuntimeError("Guest shutdown did not finish within three minutes.") + + +def reboot() -> None: + stop() + lxc("start", NAME) + wait_agent() + + +def guest_address() -> str: + deadline = time.monotonic() + 120 + while time.monotonic() < deadline: + state = query("/1.0/instances/" + NAME + "/state") + for interface in (state.get("network") or {}).values(): + for address in interface.get("addresses", []): + if address["family"] == "inet" and address["scope"] == "global": + return address["address"] + time.sleep(2) + raise RuntimeError("Guest did not receive an IPv4 address within two minutes.") + + +def ssh_command(*args: str) -> list[str]: + checked_instance() + actual_owner = output( + "lxc", + "--force-local", + "--project", + "default", + "exec", + NAME, + "--", + "cat", + "/etc/canonical-lab", + ) + if actual_owner != owner(): + raise RuntimeError("Guest marker does not match this lab.") + address = guest_address() + key = subprocess.check_output( + [ + "lxc", + "--force-local", + "--project", + "default", + "exec", + NAME, + "--", + "cat", + "/etc/ssh/ssh_host_ed25519_key.pub", + ], + text=True, + ).strip() + known_hosts = STATE / "known_hosts" + expected = NAME + " " + key + "\n" + if known_hosts.exists() and known_hosts.read_text() != expected: + raise RuntimeError("Guest SSH host key changed.") + if not known_hosts.exists(): + private_file(known_hosts, expected) + return [ + "ssh", + "-F", + "/dev/null", + "-i", + str(STATE / "secrets/ssh"), + "-o", + "IdentityAgent=none", + "-o", + "IdentitiesOnly=yes", + "-o", + "ForwardAgent=no", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=15", + "-o", + "ServerAliveInterval=30", + "-o", + "ServerAliveCountMax=6", + "-o", + "StrictHostKeyChecking=yes", + "-o", + "HostKeyAlias=" + NAME, + "-o", + "UserKnownHostsFile=" + str(known_hosts), + "canonical-test@" + address, + shlex.join(args), + ] + + +def remote(*args: str, desktop: bool = False, **kwargs): + if desktop: + args = ( + "systemd-run", + "--user", + "--wait", + "--pipe", + "--collect", + "--expand-environment=no", + *args, + ) + return run(*ssh_command(*args), **kwargs) + + +def logs() -> None: + checked_instance() + directory = STATE / "reports" / time.strftime("%Y%m%d-%H%M%S") + directory.mkdir(parents=True, mode=0o700) + commands = { + "system-journal": ["journalctl", "-b", "-p", "warning", "--no-pager"], + "units": ["systemctl", "--failed", "--no-pager"], + "storage": ["lsblk", "-f"], + "desktop": ["journalctl", "-b", "_UID=1000", "--no-pager", "-n", "300"], + } + for name, args in commands.items(): + with (directory / (name + ".log")).open("w") as stream: + guest(*args, stdout=stream, stderr=subprocess.STDOUT) + print("Reports: " + str(directory)) + + +def guest_stage(stage: str, desktop: bool = False) -> None: + if desktop: + wait_desktop() + directory = STATE / "reports" + directory.mkdir(exist_ok=True, mode=0o700) + log = directory / (time.strftime("%Y%m%d-%H%M%S-") + stage + ".log") + print("Running guest stage " + stage + "; log: " + str(log), flush=True) + code = 1 + try: + with log.open("w") as stream: + remote( + "bash", + "/home/canonical-test/dotfiles/scripts/canonical-vm-guest.sh", + stage, + desktop=desktop, + stdout=stream, + stderr=subprocess.STDOUT, + timeout=14400, + ) + code = 0 + except subprocess.CalledProcessError as error: + code = error.returncode + raise + finally: + log.with_suffix(".json").write_text( + json.dumps({"stage": stage, "exit_code": code}) + "\n" + ) + + +def wait_desktop() -> None: + deadline = time.monotonic() + 180 + while time.monotonic() < deadline: + try: + remote( + "sh", + "-c", + 'test -n "$WAYLAND_DISPLAY" ' + '&& test "${XDG_CURRENT_DESKTOP#*GNOME}" != "$XDG_CURRENT_DESKTOP" ' + "&& /usr/bin/gnome-extensions list >/dev/null " + '&& test ! -e "$XDG_RUNTIME_DIR/gnome-shell-disable-extensions"', + desktop=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=20, + ) + return + except subprocess.CalledProcessError: + time.sleep(5) + raise RuntimeError("The GNOME user session is not ready.") + + +def sync_source() -> None: + wait_agent() + archive = source_archive() + remote("mkdir", "-p", "/home/canonical-test/dotfiles") + remote("test", "!", "-L", "/home/canonical-test/dotfiles") + remote( + "find", + "/home/canonical-test/dotfiles", + "-mindepth", + "1", + "-maxdepth", + "1", + "!", + "-name", + ".git", + "-exec", + "rm", + "-rf", + "--", + "{}", + "+", + ) + with archive.open("rb") as stream: + remote("tar", "-xf", "-", "-C", "/home/canonical-test/dotfiles", stdin=stream) + + +def configure_keyring() -> None: + with (STATE / "secrets/login-password").open("rb") as stream: + guest( + "bash", + "/home/canonical-test/dotfiles/scripts/canonical-vm-keyring.sh", + stdin=stream, + ) + + +def deploy_guest() -> None: + configure_keyring() + reboot() + guest_stage("setup", desktop=True) + + +def bootstrap() -> None: + sync_source() + signing_key = STATE / "secrets/signing.gpg" + if signing_key.exists(): + with signing_key.open("rb") as stream: + remote("gpg", "--batch", "--import", stdin=stream) + guest_stage("identity") + if not signing_key.exists(): + with signing_key.open("xb") as stream: + signing_key.chmod(0o600) + remote( + "gpg", + "--batch", + "--export-secret-keys", + "canonical-test@example.invalid", + stdout=stream, + ) + guest_stage("nix") + stop() + snapshot("nix-ready") + lxc("start", NAME) + wait_agent() + deploy_guest() + + +def test_guest() -> None: + wait_agent() + wait_desktop() + guest_stage("settings", desktop=True) + reboot() + wait_desktop() + guest_stage("check", desktop=True) + screenshot() + reboot() + wait_desktop() + guest_stage("session", desktop=True) + screenshot() + logs() + + +def wait_install() -> None: + deadline = time.monotonic() + 5400 + while time.monotonic() < deadline: + if checked_instance()["status"] == "Stopped": + return + time.sleep(15) + raise RuntimeError( + "Installer has not stopped after 90 minutes. Inspect the guest console." + ) + + +def screenshot() -> None: + for attempt in range(3): + try: + capture_screenshot() + return + except subprocess.CalledProcessError as error: + if error.cmd[0] != "spicy-screenshot" or attempt == 2: + raise + print("SPICE capture failed; retrying in three seconds.", flush=True) + time.sleep(3) + + +def capture_screenshot() -> None: + checked_instance() + client = shutil.which("lxc") + if not client: + raise RuntimeError("LXD client is missing.") + directory = STATE / "reports" + directory.mkdir(exist_ok=True, mode=0o700) + target = directory / (time.strftime("%Y%m%d-%H%M%S") + ".ppm") + env = {**os.environ, "PATH": "/nonexistent"} + with subprocess.Popen( + [ + client, + "--force-local", + "--project", + "default", + "console", + NAME, + "--type=vga", + ], + env=env, + cwd=STATE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) as proxy: + try: + assert proxy.stdout + deadline = time.monotonic() + 30 + data = b"" + while time.monotonic() < deadline: + if select.select([proxy.stdout], [], [], 1)[0]: + data += os.read(proxy.stdout.fileno(), 4096) + match = re.search(rb"spice\+unix://[^\s]+", data) + if match: + run( + "spicy-screenshot", + "--uri=" + match[0].decode(), + "--out-file=" + str(target), + cwd=STATE, + env={ + **os.environ, + "SPICE_DISABLE_CHANNELS": "playback-0,record-0", + }, + timeout=60, + ) + run("magick", str(target), str(target.with_suffix(".png"))) + print(target.with_suffix(".png")) + return + if proxy.poll() is not None: + break + raise RuntimeError( + "No guest SPICE socket returned: " + data.decode(errors="replace") + ) + finally: + proxy.terminate() + try: + proxy.wait(timeout=10) + except subprocess.TimeoutExpired: + proxy.kill() + proxy.wait() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "action", + choices=[ + "create", + "status", + "console", + "start", + "stop", + "snapshot", + "restore", + "delete", + "finish-install", + "bootstrap", + "test", + "logs", + "screenshot", + "wait-install", + "reboot", + "deploy", + "sync-source", + ], + ) + parser.add_argument( + "snapshot", nargs="?", choices=["clean", "nix-ready", "working", "diagnostic"] + ) + parser.add_argument("--mirror", default=RELEASE) + args = parser.parse_args() + if args.action == "create": + if not args.mirror.startswith("https://"): + parser.error("Use an HTTPS ISO mirror.") + create(args.mirror) + return + instance = checked_instance() + if args.action == "bootstrap": + bootstrap() + elif args.action == "sync-source": + sync_source() + guest_stage("identity") + elif args.action == "wait-install": + wait_install() + elif args.action == "deploy": + deploy_guest() + elif args.action == "reboot": + reboot() + elif args.action == "stop": + stop() + elif args.action == "test": + test_guest() + elif args.action == "logs": + logs() + elif args.action == "screenshot": + screenshot() + elif args.action == "status": + print(json.dumps(instance, indent=2)) + elif args.action == "console": + lxc("console", NAME, "--type=vga", cwd=STATE) + elif args.action == "snapshot": + if not args.snapshot: + parser.error("snapshot name required") + snapshot(args.snapshot) + elif args.action == "restore": + if not args.snapshot: + parser.error("snapshot name required") + if instance["status"] != "Stopped": + raise RuntimeError("Stop the lab VM before restore.") + lxc("restore", NAME, args.snapshot) + elif args.action == "finish-install": + if instance["status"] != "Stopped": + raise RuntimeError("Wait for the installer to power off.") + lxc("config", "device", "remove", NAME, "installer") + lxc("start", NAME) + wait_agent() + guest("test", "-f", "/etc/canonical-lab") + stop() + snapshot("clean") + else: + lxc(args.action, NAME) + + +if __name__ == "__main__": + main() diff --git a/tests/test_canonical.py b/tests/test_canonical.py index b21b8f5..e1c825c 100644 --- a/tests/test_canonical.py +++ b/tests/test_canonical.py @@ -17,12 +17,52 @@ SPEC.loader.exec_module(canonical) class PackageTests(unittest.TestCase): + def test_install_does_not_enable_experimental_snap_features(self): + with patch.object(canonical.subprocess, "run") as command: + canonical.install() + self.assertFalse(any("set" in call.args[0] for call in command.call_args_list)) + def test_keybase_is_not_installed_or_autostarted(self): self.assertNotIn("keybase", canonical.packages("snap")) self.assertFalse( (ROOT / "dot_config/autostart/dotfiles-keybase.desktop").exists() ) + def test_extensions_install_without_shell_confirmation(self): + with ( + patch("sys.argv", ["canonical.py", "extensions"]), + patch.object(canonical, "require_canonical"), + patch.object(canonical.subprocess, "run") as command, + ): + canonical.main() + self.assertEqual( + command.call_args.args[0][:3], ["gext", "--filesystem", "install"] + ) + + def test_lab_check_rejects_missing_marker(self): + with ( + patch.object(canonical.Path, "is_file", return_value=False), + self.assertRaises(SystemExit), + ): + canonical.require_lab() + + def test_normal_check_keeps_company_registration(self): + with patch.object(canonical.subprocess, "run") as command: + command.return_value.returncode = 0 + canonical.check() + self.assertTrue( + any( + call.args[0] == ["landscape-config", "--actively-registered"] + for call in command.call_args_list + ) + ) + self.assertTrue( + any( + call.args[0] == ["nix", "store", "ping", "--store", "daemon"] + for call in command.call_args_list + ) + ) + def test_non_corporate_role_is_rejected_before_system_access(self): with ( patch.object( diff --git a/tests/test_canonical_vm.py b/tests/test_canonical_vm.py new file mode 100644 index 0000000..481242b --- /dev/null +++ b/tests/test_canonical_vm.py @@ -0,0 +1,230 @@ +import importlib.util +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location( + "canonical_vm", ROOT / "scripts/canonical_vm.py" +) +assert SPEC and SPEC.loader +vm = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(vm) + + +class LabTests(unittest.TestCase): + def test_reboot_checks_extensions_without_reapplying_settings(self): + with ( + patch.object(vm, "wait_agent"), + patch.object(vm, "wait_desktop"), + patch.object(vm, "guest_stage") as stage, + patch.object(vm, "reboot"), + patch.object(vm, "remote"), + patch.object(vm, "screenshot"), + patch.object(vm, "logs"), + ): + vm.test_guest() + self.assertEqual(stage.call_args.args, ("session",)) + + def test_desktop_waits_for_extension_startup_check(self): + with patch.object(vm, "remote") as remote: + vm.wait_desktop() + command = remote.call_args.args[2] + self.assertIn("gnome-extensions list", command) + self.assertIn( + 'test ! -e "$XDG_RUNTIME_DIR/gnome-shell-disable-extensions"', command + ) + + def test_screenshot_retries_a_failed_spice_connection(self): + error = subprocess.CalledProcessError(1, ["spicy-screenshot"]) + with ( + patch.object( + vm, "capture_screenshot", side_effect=[error, None] + ) as capture, + patch.object(vm.time, "sleep"), + ): + vm.screenshot() + self.assertEqual(capture.call_count, 2) + + def test_screenshot_retry_is_bounded(self): + error = subprocess.CalledProcessError(1, ["spicy-screenshot"]) + with ( + patch.object(vm, "capture_screenshot", side_effect=error) as capture, + patch.object(vm.time, "sleep"), + self.assertRaises(subprocess.CalledProcessError), + ): + vm.screenshot() + self.assertEqual(capture.call_count, 3) + + def test_deploy_prepares_keyring_before_desktop_setup(self): + calls = [] + with ( + patch.object( + vm, "configure_keyring", side_effect=lambda: calls.append("keyring") + ), + patch.object(vm, "reboot", side_effect=lambda: calls.append("reboot")), + patch.object( + vm, "guest_stage", side_effect=lambda *a, **k: calls.append("setup") + ), + ): + vm.deploy_guest() + self.assertEqual(calls, ["keyring", "reboot", "setup"]) + + def test_keyring_password_is_sent_on_stdin(self): + with tempfile.TemporaryDirectory() as directory: + state = Path(directory) + (state / "secrets").mkdir() + (state / "secrets/login-password").write_bytes(b"test-password") + with ( + patch.object(vm, "STATE", state), + patch.object(vm, "guest") as guest, + ): + guest.side_effect = lambda *args, **kwargs: self.assertEqual( + kwargs["stdin"].read(), b"test-password" + ) + vm.configure_keyring() + self.assertNotIn("test-password", str(guest.call_args.args)) + + def test_address_waits_for_dhcp(self): + ready = { + "network": { + "enp5s0": { + "addresses": [ + {"family": "inet", "scope": "global", "address": "192.0.2.10"} + ] + } + } + } + with ( + patch.object(vm, "query", side_effect=[{"network": {}}, ready]), + patch.object(vm.time, "sleep") as sleep, + ): + self.assertEqual(vm.guest_address(), "192.0.2.10") + sleep.assert_called_once() + + def test_stop_uses_guest_shutdown_instead_of_power_button(self): + with ( + patch.object( + vm, + "checked_instance", + side_effect=[{"status": "Running"}, {"status": "Stopped"}], + ), + patch.object(vm, "guest") as guest, + patch.object(vm, "lxc") as lxc, + ): + vm.stop() + guest.assert_called_once_with("systemctl", "poweroff", "--no-block") + lxc.assert_not_called() + + def test_desktop_commands_preserve_shell_expansion(self): + with ( + patch.object(vm, "ssh_command", side_effect=lambda *args: ["ssh", *args]), + patch.object(vm, "run") as command, + ): + vm.remote("sh", "-c", 'test -n "$WAYLAND_DISPLAY"', desktop=True) + self.assertIn("--expand-environment=no", command.call_args.args) + + def test_source_checks_exclude_lab_data(self): + with tempfile.TemporaryDirectory() as directory: + base = Path(directory) + (base / "tmp").mkdir() + (base / "tmp/generated.py").touch() + (base / "source.py").touch() + result = subprocess.check_output( + [ + "bash", + "-c", + 'source "$1"; _find_by_ext py', + "bash", + str(ROOT / "just-lib.sh"), + ], + cwd=base, + text=True, + ) + self.assertEqual(result.strip(), "./source.py") + + def test_ownership_requires_vm_and_matching_marker(self): + good = {"type": "virtual-machine", "config": {vm.MARKER: "test-id"}} + vm.require_owned(good, "test-id") + for bad in [{}, {**good, "type": "container"}, {**good, "config": {}}]: + with self.assertRaises(RuntimeError): + vm.require_owned(bad, "test-id") + + def test_source_filter_excludes_secrets_and_host_git(self): + for path in ["tmp/key", ".git/config", ".direnv/env", "x/__pycache__/x.pyc"]: + self.assertFalse(vm.source_allowed(path)) + self.assertTrue(vm.source_allowed("dot_config/git/config.tmpl")) + self.assertTrue(vm.source_allowed("scripts/canonical_vm.py")) + + def test_autoinstall_targets_only_lab_disk(self): + config = vm.autoinstall("hash", "ssh-ed25519 test", "disk-secret") + install = config["autoinstall"] + self.assertEqual(install["storage"]["layout"]["match"], {"path": "/dev/sda"}) + self.assertEqual(install["storage"]["layout"]["name"], "lvm") + self.assertFalse(install["ssh"]["allow-pw"]) + self.assertNotIn("landscape", str(config)) + self.assertNotIn("interactive-sections", install) + self.assertEqual( + install["early-commands"][0], ["systemd-detect-virt", "--vm", "--quiet"] + ) + + def test_installer_boot_patch_requires_kernel(self): + result = vm.boot_config("set timeout=30\n linux /casper/vmlinuz quiet ---\n") + self.assertIn("autoinstall", result) + self.assertIn("set timeout=1", result) + with self.assertRaises(RuntimeError): + vm.boot_config("unknown installer") + + def test_snapshot_rejects_running_vm(self): + with ( + patch.object(vm, "checked_instance", return_value={"status": "Running"}), + patch.object(vm, "lxc") as command, + ): + with self.assertRaises(RuntimeError): + vm.snapshot("clean") + command.assert_not_called() + + def test_ssh_does_not_use_host_config_or_agent(self): + state = { + "network": { + "enp5s0": { + "addresses": [ + {"address": "192.0.2.10", "family": "inet", "scope": "global"} + ] + } + } + } + with ( + tempfile.TemporaryDirectory() as directory, + patch.object(vm, "STATE", Path(directory)), + patch.object(vm, "checked_instance"), + patch.object(vm, "owner", return_value="test-id"), + patch.object(vm, "output", return_value="test-id"), + patch.object(vm, "query", return_value=state), + patch.object( + vm.subprocess, "check_output", return_value="ssh-ed25519 example" + ), + ): + command = vm.ssh_command("true") + self.assertEqual(command[:3], ["ssh", "-F", "/dev/null"]) + for option in [ + "IdentityAgent=none", + "ForwardAgent=no", + "StrictHostKeyChecking=yes", + "IdentitiesOnly=yes", + ]: + self.assertIn(option, command) + (Path(directory) / "known_hosts").write_text("changed") + with self.assertRaises(RuntimeError): + vm.ssh_command("true") + + def test_private_file_does_not_replace_existing_secret(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "secret" + vm.private_file(path, "first") + with self.assertRaises(FileExistsError): + vm.private_file(path, "second") + self.assertEqual(path.read_text(), "first") + self.assertEqual(path.stat().st_mode & 0o777, 0o600) |