diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 822f841..9dd8403 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -117,6 +117,14 @@ WinPE; the Blancco and Clonezilla utilities require Secure Boot disabled. The menu checks the CPU (iPXE cpuid --ext 29): 64-bit CPUs get the current release, 32-bit-only CPUs (XP-era Pentium 4/M, early Atom) get clonezilla-i686/ (3.2.0-5, the last 32-bit Clonezilla) over legacy BIOS. + "Clonezilla: back up / restore this PC" ask for the machine number, + mount \\\clonezilla as the clonezilla Samba user + (ocs_repository smb3://, password from clonezilla_smb_password) and + run clonezilla-scripts/pxe-cz.sh (ocs_prerun downloads it, + ocs_live_run runs it): images are - with + pxe-meta.txt, checked after saving; restore lists that machine's + images and restores with -k1 -r (larger disks get proportionally + larger partitions). "Clonezilla: manual" is the interactive mode. Memory testing is not on the menu: Dell PCs have a built-in test (F12 at power-on, Diagnostics). Memtest86+ was removed because it is diff --git a/playbook/clonezilla-scripts/pxe-cz.sh b/playbook/clonezilla-scripts/pxe-cz.sh new file mode 100644 index 0000000..d404893 --- /dev/null +++ b/playbook/clonezilla-scripts/pxe-cz.sh @@ -0,0 +1,473 @@ +#!/bin/bash +# PXE Clonezilla: back up or restore a PC by its machine number. +# +# The iPXE menu (GetPxeScript.aspx, "Clonezilla: back up / restore this +# PC") boots Clonezilla live with: +# ocs_prerun="wget ... -O /tmp/pxe-cz.sh http:///clonezilla-scripts/pxe-cz.sh" +# ocs_repository="smb3://clonezilla:@/clonezilla/" +# ocs_live_run="bash /tmp/pxe-cz.sh" +# pxe_cz_mode=save|restore pxe_machine= pxe_end=1 +# Clonezilla runs the prerun (this download), mounts the repository on +# /home/partimag, then runs this script as root on the console. +# +# Images are named - (server time, site zone), +# with pxe-meta.txt (machine, serial, model, disk, date, Clonezilla +# version, 32/64-bit, check result) next to Clonezilla's own files. The +# PXE webapp (Tools > Clonezilla backups) reads both. +# +# Runs under Clonezilla live's bash (3.3.3 amd64 and 3.2.0-5 i686, both +# Debian with coreutils, util-linux, dmidecode, wget). Plain ASCII only. + +set -u + +# PXE_CZ_* overrides exist only for testing this script off Clonezilla. +OCSROOT="${PXE_CZ_OCSROOT:-/home/partimag}" +CMDLINE_FILE="${PXE_CZ_CMDLINE:-/proc/cmdline}" +SYS_BLOCK="${PXE_CZ_SYS_BLOCK:-/sys/block}" +SITE_TZ="America/New_York" +SERVER="$(sed -n 's/.*ocs_prerun="[^"]*http:\/\/\([^/":]*\).*/\1/p' "$CMDLINE_FILE" 2>/dev/null)" +[ -n "$SERVER" ] || SERVER=172.16.9.1 +CMDLINE=" $(cat "$CMDLINE_FILE" 2>/dev/null) " + +# --- output helpers ------------------------------------------------------- + +line() { echo "======================================================================"; } + +banner() { + echo + line + for l in "$@"; do echo " $l"; done + line + echo +} + +# Ask what to do at the end. Success: power off after 60 s unless a key is +# pressed. Failure: wait, the operator has to read the message. +finish() { + local ok="$1" ans + sync + while true; do + echo + echo " [Enter] power off r = reboot s = shell (type exit to come back)" + if [ "$ok" = "yes" ]; then + echo -n " Powering off in 60 seconds... " + if ! read -r -t 60 ans; then ans=""; echo; fi + else + echo -n " Choice: " + read -r ans || ans="" + fi + case "$ans" in + r|R) umount "$OCSROOT" 2>/dev/null; sync; reboot; sleep 30 ;; + s|S) bash -i ;; + *) umount "$OCSROOT" 2>/dev/null; sync; poweroff; sleep 30 ;; + esac + done +} + +fail() { + banner "FAILED: $1" "${@:2}" + finish no + exit 1 +} + +# --- machine number ------------------------------------------------------- + +# Letters, digits and dash, starting with a letter or digit, up to 32. +valid_machine() { + [[ "$1" =~ ^[A-Za-z0-9][A-Za-z0-9-]{0,31}$ ]] +} + +# The text typed at the iPXE prompt sits between pxe_machine= and the +# pxe_end=1 sentinel, so a typed space shows up here and is rejected +# instead of being split into another kernel parameter. +machine_from_cmdline() { + local rest + case "$CMDLINE" in + *" pxe_machine="*" pxe_end=1 "*) ;; + *) echo ""; return ;; + esac + rest="${CMDLINE#* pxe_machine=}" + echo "${rest%% pxe_end=1 *}" +} + +# Sets MACHINE (upper case). Not run in $(...): finish() reads the +# keyboard and must write to the console. +get_machine() { + local m + m="$(machine_from_cmdline)" + if [ -z "$m" ] || ! valid_machine "$m"; then + if [ -n "$m" ]; then + banner "The machine number typed at the boot menu is not valid:" \ + " [$m]" "Use letters, digits and dash only (for example 6501 or WJ-6501)." + else + banner "No machine number was given at the boot menu." + fi + while true; do + echo -n " Machine number (letters, digits, dash; empty = stop): " + read -r m || m="" + [ -n "$m" ] || finish no + valid_machine "$m" && break + echo " [$m] is not a valid machine number: use letters, digits and dash only." + done + fi + MACHINE="$(echo "$m" | tr '[:lower:]' '[:upper:]')" +} + +# --- time ----------------------------------------------------------------- + +# The PC's clock is often wrong (dead CMOS battery on old PCs, local time +# on Windows PCs), so take the time from the PXE server's HTTP Date header +# and set the live system clock from it. Falls back to the PC clock. +TIME_SOURCE="pc-clock" +sync_clock() { + local hdr epoch + hdr="$(wget -q -S --spider -T 10 -t 2 "http://$SERVER/clonezilla-scripts/pxe-cz.sh" 2>&1 \ + | sed -n 's/^ *Date: *//p' | head -1)" + [ -n "$hdr" ] || return 0 + epoch="$(date -u -d "$hdr" +%s 2>/dev/null)" || return 0 + [ -n "$epoch" ] || return 0 + date -u -s "@$epoch" >/dev/null 2>&1 && TIME_SOURCE="pxe-server" +} + +site_date() { + if [ -e "/usr/share/zoneinfo/$SITE_TZ" ]; then + TZ="$SITE_TZ" date "$@" + else + date "$@" + fi +} + +# --- hardware ------------------------------------------------------------- + +dmi() { + local v + v="$(dmidecode -s "$1" 2>/dev/null | grep -v '^#' | head -1 | tr -d '\r' | sed 's/^ *//; s/ *$//')" + echo "$v" +} + +# Whole disks a backup can come from or a restore can go to: no CD, loop, +# RAM, zram, device-mapper, md, eMMC boot areas, or empty card readers. +list_disks() { + local d n size + for d in "$SYS_BLOCK"/*; do + n="${d##*/}" + case "$n" in + loop*|sr*|ram*|zram*|dm-*|md*|fd*|nbd*|*boot[0-9]|*rpmb) continue ;; + esac + size="$(cat "$d/size" 2>/dev/null)" + [ -n "$size" ] && [ "$size" -gt 0 ] 2>/dev/null || continue + echo "$n" + done +} + +disk_bytes() { echo $(( $(cat "$SYS_BLOCK/$1/size" 2>/dev/null || echo 0) * 512 )); } + +disk_model() { + local m + m="$(cat "$SYS_BLOCK/$1/device/model" 2>/dev/null | sed 's/^ *//; s/ *$//')" + [ -n "$m" ] || m="$(lsblk -dn -o MODEL "/dev/$1" 2>/dev/null | sed 's/^ *//; s/ *$//')" + [ -n "$m" ] || m="unknown model" + echo "$m" +} + +disk_tran() { lsblk -dn -o TRAN "/dev/$1" 2>/dev/null | sed 's/ //g'; } + +human() { + local b="$1" + awk -v b="$b" 'BEGIN { if (b >= 1e12) printf "%.1f TB", b/1e12; else printf "%.0f GB", b/1e9 }' +} + +describe_disk() { + local n="$1" t + t="$(disk_tran "$n")" + echo "/dev/$n $(human "$(disk_bytes "$n")") $(disk_model "$n")${t:+ ($t)}" +} + +# pick_disk "prompt" -> disk name on stdout. One disk: that one. +pick_disk() { + local prompt="$1" disks n i ans + mapfile -t disks < <(list_disks) + if [ "${#disks[@]}" -eq 0 ]; then + echo "" + return + fi + if [ "${#disks[@]}" -eq 1 ]; then + echo "${disks[0]}" + return + fi + while true; do + { + echo + echo " $prompt" + i=1 + for n in "${disks[@]}"; do + echo " $i) $(describe_disk "$n")" + i=$((i + 1)) + done + echo -n " Number (1-${#disks[@]}): " + } >&2 + read -r ans || ans="" + if [[ "$ans" =~ ^[0-9]+$ ]] && [ "$ans" -ge 1 ] && [ "$ans" -le "${#disks[@]}" ]; then + echo "${disks[$((ans - 1))]}" + return + fi + echo " Type a number from the list." >&2 + done +} + +live_version() { + local v="" + if [ -e /etc/ocs/ocs-live.conf ]; then + # shellcheck disable=SC1091 + v="$(. /etc/ocs/ocs-live.conf >/dev/null 2>&1; echo "${drbl_ocs_live_ver:-}")" + fi + [ -n "$v" ] || v="clonezilla $(dpkg-query -W -f='${Version}' clonezilla 2>/dev/null)" + echo "$v" +} + +bits() { + case "$(uname -m)" in + x86_64|amd64) echo 64 ;; + *) echo 32 ;; + esac +} + +# set_meta FILE KEY VALUE: replace or add one key=value line. +set_meta() { + local f="$1" k="$2" v="$3" tmp + v="$(echo "$v" | tr -d '\r\n')" + tmp="$(mktemp /tmp/pxe-meta.XXXXXX)" + [ -e "$f" ] && grep -v "^$k=" "$f" > "$tmp" + echo "$k=$v" >> "$tmp" + cat "$tmp" > "$f" + rm -f "$tmp" +} + +get_meta() { + sed -n "s/^$2=//p" "$1" 2>/dev/null | head -1 +} + +repo_ready() { + mountpoint -q "$OCSROOT" 2>/dev/null || grep -qs " $OCSROOT cifs " /proc/mounts +} + +# --- backup --------------------------------------------------------------- + +do_save() { + local m="$1" disk stamp img n meta rc chk + disk="$(pick_disk "This PC has more than one disk. Which one do you want to back up?")" + [ -n "$disk" ] || fail "No disk found to back up." "Check the disk is connected and shows in the BIOS." + stamp="$(site_date +%Y-%m-%d-%H%M)" + img="$m-$stamp" + n=2 + while [ -e "$OCSROOT/$img" ]; do + img="$m-$stamp-$n" + n=$((n + 1)) + done + + banner "BACKUP of machine $m" \ + "Disk: $(describe_disk "$disk")" \ + "Image: $img" \ + "To: \\\\$SERVER\\clonezilla\\$img" + + ocs-sr -q2 -batch -j2 -z9p -i 0 -sfsck -scs -senc -p command savedisk "$img" "$disk" + rc=$? + + meta="$OCSROOT/$img/pxe-meta.txt" + if [ ! -d "$OCSROOT/$img" ]; then + fail "Clonezilla did not create the image $img (exit code $rc)." \ + "See the messages above. Nothing was saved." + fi + : > "$meta" + set_meta "$meta" machine "$m" + set_meta "$meta" date "$(site_date '+%Y-%m-%d %H:%M %Z')" + set_meta "$meta" date_utc "$(date -u +%Y-%m-%dT%H:%M:%SZ)" + set_meta "$meta" time_source "$TIME_SOURCE" + set_meta "$meta" serial "$(dmi system-serial-number)" + set_meta "$meta" manufacturer "$(dmi system-manufacturer)" + set_meta "$meta" model "$(dmi system-product-name)" + set_meta "$meta" disk "$disk" + set_meta "$meta" disk_model "$(disk_model "$disk")" + set_meta "$meta" disk_size_bytes "$(disk_bytes "$disk")" + set_meta "$meta" clonezilla "$(live_version)" + set_meta "$meta" bits "$(bits)" + if [ -d /sys/firmware/efi ]; then set_meta "$meta" firmware uefi; else set_meta "$meta" firmware bios; fi + if [ "$rc" -ne 0 ]; then + set_meta "$meta" save failed + set_meta "$meta" check not-run + fail "Clonezilla could not save machine $m (exit code $rc)." \ + "The partial image $img is marked as failed on the share;" \ + "delete it in the webapp (Tools > Clonezilla backups) and try again." + fi + set_meta "$meta" save ok + + banner "Checking the saved image $img ..." + ocs-chkimg -b "$img" + chk=$? + set_meta "$meta" check_date "$(site_date '+%Y-%m-%d %H:%M %Z')" + if [ "$chk" -ne 0 ]; then + set_meta "$meta" check failed + fail "The image $img was saved but its CHECK FAILED (exit code $chk)." \ + "Do not rely on it. Back this PC up again." + fi + set_meta "$meta" check ok + sync + banner "SUCCESS: machine $m backed up and checked." \ + "Image: $img" \ + "Serial: $(get_meta "$meta" serial) Model: $(get_meta "$meta" model)" + finish yes +} + +# --- restore -------------------------------------------------------------- + +is_cz_image() { + local d="$1" + [ -d "$d" ] || return 1 + [ -f "$d/disk" ] || [ -f "$d/parts" ] || return 1 + local f + for f in "$d"/*; do + case "${f##*/}" in + *-ptcl-img*|*.dd-img*|*-img.*|clonezilla-img) return 0 ;; + esac + done + return 1 +} + +# Images of one machine, newest first: -YYYY-MM-DD-HHMM[-n] by +# name, plus any image whose pxe-meta.txt says machine=. +machine_images() { + local m="$1" d n mm + for d in "$OCSROOT"/*/; do + d="${d%/}" + n="${d##*/}" + is_cz_image "$d" || continue + mm="$(get_meta "$d/pxe-meta.txt" machine | tr '[:lower:]' '[:upper:]')" + if [ "$mm" = "$m" ]; then + echo "$n" + elif [ -z "$mm" ] && echo "$n" | tr '[:lower:]' '[:upper:]' | grep -qE "^$m-[0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{4}(-[0-9]+)?$"; then + echo "$n" + fi + done | sort_newest +} + +# Sort names by their date stamp (the last YYYY-MM-DD-HHMM[-n]), newest first. +sort_newest() { + local n key + while read -r n; do + key="$(echo "$n" | grep -oE '[0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{4}(-[0-9]+)?$')" + echo "${key:-0000}|$n" + done | sort -r | cut -d'|' -f2- +} + +image_disks() { tr -s ' \t' '\n' < "$OCSROOT/$1/disk" 2>/dev/null | grep -v '^$'; } + +image_disk_bytes() { + local img="$1" src="$2" b sec + b="$(get_meta "$OCSROOT/$img/pxe-meta.txt" disk_size_bytes)" + if [[ "$b" =~ ^[0-9]+$ ]]; then echo "$b"; return; fi + # Clonezilla's sfdisk dump: "last-lba: N" (GPT) or the partitions' end. + sec="$(sed -n 's/^last-lba: *//p' "$OCSROOT/$img/$src-pt.sf" 2>/dev/null | head -1)" + if [[ "$sec" =~ ^[0-9]+$ ]]; then echo $(( (sec + 34) * 512 )); return; fi + echo 0 +} + +do_restore() { + local m="$1" imgs i n ans img src srcb tgt tgtb meta chkopt rc mo se ck + mapfile -t imgs < <(machine_images "$m") + if [ "${#imgs[@]}" -eq 0 ]; then + fail "No images for machine $m on \\\\$SERVER\\clonezilla." \ + "Check the machine number, or look in the webapp (Tools > Clonezilla backups)." + fi + while true; do + echo + echo " Images of machine $m (newest first):" + i=1 + for n in "${imgs[@]}"; do + meta="$OCSROOT/$n/pxe-meta.txt" + mo="$(get_meta "$meta" model)"; se="$(get_meta "$meta" serial)"; ck="$(get_meta "$meta" check)" + echo " $i) $n ${mo:-model ?} serial ${se:-?} check: ${ck:-?}" + i=$((i + 1)) + done + echo -n " Image to restore (1-${#imgs[@]}, q = quit): " + read -r ans || ans=q + case "$ans" in q|Q) finish no ;; esac + if [[ "$ans" =~ ^[0-9]+$ ]] && [ "$ans" -ge 1 ] && [ "$ans" -le "${#imgs[@]}" ]; then + img="${imgs[$((ans - 1))]}" + break + fi + echo " Type a number from the list." + done + meta="$OCSROOT/$img/pxe-meta.txt" + if [ "$(get_meta "$meta" save)" = "failed" ]; then + fail "The image $img is marked as a FAILED backup and cannot be restored." + fi + mapfile -t src < <(image_disks "$img") + if [ "${#src[@]}" -ne 1 ]; then + fail "The image $img holds ${#src[@]} disks (${src[*]:-none})." \ + "Restore it with Clonezilla: manual (boot menu)." + fi + srcb="$(image_disk_bytes "$img" "${src[0]}")" + + tgt="$(pick_disk "Which disk do you want to restore $img to? (ALL DATA ON IT WILL BE ERASED)")" + [ -n "$tgt" ] || fail "No disk found to restore to." + tgtb="$(disk_bytes "$tgt")" + if [ "$srcb" -gt 0 ] && [ "$tgtb" -lt "$srcb" ]; then + fail "The target disk is SMALLER than the disk the image came from." \ + "Image disk: $(human "$srcb")" "Target disk: $(describe_disk "$tgt")" \ + "Use a disk at least as large, or Clonezilla: manual (expert mode, -icds)." + fi + + banner "RESTORE machine $m" \ + "Image: $img (saved $(get_meta "$meta" date))" \ + "From: $(get_meta "$meta" model) serial $(get_meta "$meta" serial) disk $(human "$srcb")" \ + "Target: $(describe_disk "$tgt")" \ + "" \ + "EVERYTHING on /dev/$tgt will be ERASED and replaced by this image." \ + "This PC: $(dmi system-product-name) serial $(dmi system-serial-number)" + echo -n " Type YES (capitals) to erase /dev/$tgt and restore: " + read -r ans || ans="" + if [ "$ans" != "YES" ]; then + banner "Cancelled. Nothing was written." + finish no + fi + + # A larger target: -k1 recreates the partition table proportionally + # (every partition scaled by target/source size) and -r grows each + # filesystem to its new partition, so the extra space is used. The + # default (-k0) copies the table as-is and leaves the rest unallocated. + # The image was checked after it was saved; check again only if that + # check did not pass. + chkopt="" + [ "$(get_meta "$meta" check)" = "ok" ] && chkopt="-scr" + ocs-sr -g auto -e1 auto -e2 -r -j2 -k1 $chkopt -batch -p command restoredisk "$img" "$tgt" + rc=$? + if [ "$rc" -ne 0 ]; then + fail "Clonezilla could not restore $img to /dev/$tgt (exit code $rc)." \ + "The disk may be partly written: do not boot from it until a restore succeeds." + fi + sync + banner "SUCCESS: $img restored to /dev/$tgt." "At the next start, boot this PC from its disk (not the network)." + finish yes +} + +# --- main ----------------------------------------------------------------- + +MODE="$(echo "$CMDLINE" | sed -n 's/.* pxe_cz_mode=\([a-z]*\) .*/\1/p')" +case "$MODE" in + save|restore) ;; + *) fail "Unknown mode [$MODE] on the kernel command line (pxe_cz_mode=save or restore)." ;; +esac + +if ! repo_ready; then + fail "The backup share \\\\$SERVER\\clonezilla is not mounted on $OCSROOT." \ + "Check the PXE server is up and the clonezilla Samba account works" \ + "(playbook var clonezilla_smb_password), then boot this option again." +fi + +sync_clock +MACHINE="" +get_machine +case "$MODE" in + save) do_save "$MACHINE" ;; + restore) do_restore "$MACHINE" ;; +esac diff --git a/playbook/ipxe/GetPxeScript.aspx b/playbook/ipxe/GetPxeScript.aspx index d3d731e..2c38988 100644 --- a/playbook/ipxe/GetPxeScript.aspx +++ b/playbook/ipxe/GetPxeScript.aspx @@ -10,8 +10,10 @@ item --gap -- ---- Drive Erase ---- item blancco Blancco Drive Eraser item blancco_debug Blancco Drive Eraser (debug / SSH) item --gap -- ---- Utilities ---- -item clonezilla Clonezilla Live (Disk Imaging) -item clonezilla32 Clonezilla Live 32-bit (XP-era PCs, forced) +item cz_backup Clonezilla: back up this PC +item cz_restore Clonezilla: restore this PC +item clonezilla Clonezilla: manual (USB or other location) +item clonezilla32 Clonezilla: manual, 32-bit (XP-era PCs, forced) item --gap -- Memory test: Dell PCs press F12 at power-on, Diagnostics item --gap -- ---- item reboot Reboot @@ -48,6 +50,47 @@ echo Windows PE boot FAILED (see the error above). echo Check that the PXE server is up and the win11 boot files are staged. prompt Press any key to return to menu... && goto menu || goto menu +# --- Clonezilla by machine number (back up / restore) --------------------- +# The image share is \\${server}\clonezilla (Samba account clonezilla, +# the playbook fills in its password). Clonezilla mounts it on +# /home/partimag (ocs_repository, SMB 3), downloads +# clonezilla-scripts/pxe-cz.sh (ocs_prerun) and runs it (ocs_live_run): +# it checks the machine number, saves - or lists +# that machine's images to restore, and writes pxe-meta.txt. +:cz_backup +set cz_mode save +goto cz_machine + +:cz_restore +set cz_mode restore +goto cz_machine + +:cz_machine +# iPXE cannot check characters: pxe-cz.sh accepts letters, digits and dash +# and asks again there. Empty: ask again. Esc / Ctrl-C: back to the menu. +echo +echo -n Machine number: && read czmachine || goto menu +isset ${czmachine} || goto cz_machine +# Same CPU check as the manual entry below (cpuid --ext 29 = 64-bit). +cpuid --ext 29 && goto cz_auto64 || goto cz_auto32 + +:cz_auto64 +set base http://${server}/clonezilla +kernel ${base}/vmlinuz boot=live username=user union=overlay config components noswap edd=on nomodeset nodmraid locales=en_US.UTF-8 keyboard-layouts=us ocs_prerun="wget -q -T 15 -t 5 -O /tmp/pxe-cz.sh http://${server}/clonezilla-scripts/pxe-cz.sh" ocs_repository="smb3://clonezilla:__CLONEZILLA_SMB_PASSWORD__@${server}/clonezilla/" echo_ocs_repository=no ocs_live_run="bash /tmp/pxe-cz.sh" ocs_live_extra_param="" ocs_live_batch=no net.ifnames=0 nosplash noprompt fetch=${base}/filesystem.squashfs pxe_cz_mode=${cz_mode} pxe_machine=${czmachine} pxe_end=1 || goto secureboot_warn +initrd ${base}/initrd.img || goto boot_failed +# Secure Boot ON: Debian's shim, as in :clonezilla64 below. +shim ${base}/shimx64.efi || +boot || goto secureboot_warn + +:cz_auto32 +# 3.2.0-5 i686 over legacy BIOS only; UEFI goes to the 64-bit build. +iseq ${platform} efi && goto cz_auto64 || +set base http://${server}/clonezilla-i686 +kernel ${base}/vmlinuz boot=live username=user union=overlay config components noswap edd=on nomodeset nodmraid locales=en_US.UTF-8 keyboard-layouts=us ocs_prerun="wget -q -T 15 -t 5 -O /tmp/pxe-cz.sh http://${server}/clonezilla-scripts/pxe-cz.sh" ocs_repository="smb3://clonezilla:__CLONEZILLA_SMB_PASSWORD__@${server}/clonezilla/" echo_ocs_repository=no ocs_live_run="bash /tmp/pxe-cz.sh" ocs_live_extra_param="" ocs_live_batch=no net.ifnames=0 nosplash noprompt fetch=${base}/filesystem.squashfs pxe_cz_mode=${cz_mode} pxe_machine=${czmachine} pxe_end=1 || goto boot_failed +initrd ${base}/initrd.img || goto boot_failed +boot || goto boot_failed + +# --- Clonezilla manual (interactive: USB, SSH, NFS or any share) ------------ :clonezilla # 64-bit CPUs (cpuid extended bit 29 = long mode) get the current Clonezilla; # older 32-bit-only CPUs (Pentium 4 without EM64T, Pentium M, early Atom: diff --git a/playbook/pxe_server_setup.yml b/playbook/pxe_server_setup.yml index d46ddc6..20e4bf3 100644 --- a/playbook/pxe_server_setup.yml +++ b/playbook/pxe_server_setup.yml @@ -180,6 +180,14 @@ # suffix is not blancco_ksuffix. blancco_kver: "7.0.0-34-generic" blancco_ksuffix: "{{ blancco_kver.split('.')[0] }}{{ blancco_kver.split('.')[1] }}" + # Samba account for the [clonezilla] share only (Clonezilla back up / + # restore by machine number). The iPXE menu carries it in Clonezilla's + # ocs_repository URI (smb3://clonezilla:@server/clonezilla/), + # so any imaging client can read it: it opens that one share and + # nothing else. Set a site value in site-vars.yml. Allowed characters: + # letters, digits . _ ~ - (it sits in a URI, a kernel command line and + # a mount option list), 8 to 64 of them. + clonezilla_smb_password: "clonezilla-backup" image_types: - gea-standard - gea-engineer @@ -250,6 +258,17 @@ Fix playbook/site-vars.yml and re-run. quiet: yes + - name: "Clonezilla: check the clonezilla Samba password" + assert: + that: + - clonezilla_smb_password is string + - clonezilla_smb_password is match('^[A-Za-z0-9._~-]{8,64}$') + fail_msg: >- + clonezilla_smb_password must be 8-64 characters from A-Z a-z 0-9 . _ ~ - + (it goes into Clonezilla's smb3:// URI on the kernel command line). + Fix it in playbook/site-vars.yml and re-run. + quiet: yes + # --- Re-run safety (nothing below changes the box until these pass) --- - name: "Re-run safety: is imaging in progress? (SMB sessions, recent DHCP leases)" args: @@ -754,13 +773,50 @@ # (iPXE shim command) so it runs with Secure Boot on. The Blancco # Releases page edits this file (test entry), so keep exactly one # "item blancco" line and the :blancco_efi kernel+initrd block. + # The Clonezilla back up / restore entries carry the clonezilla + # Samba password: the tracked file has the placeholder + # __CLONEZILLA_SMB_PASSWORD__, filled in here (content, so a re-run + # with the same password changes nothing). copy: - src: "{{ usb_mount }}/ipxe/GetPxeScript.aspx" + content: "{{ lookup('file', usb_mount ~ '/ipxe/GetPxeScript.aspx', rstrip=False) | replace('__CLONEZILLA_SMB_PASSWORD__', clonezilla_smb_password) }}" dest: "{{ web_root }}/Altiris/iPXE/GetPxeScript.aspx" mode: '0644' backup: yes + diff: false # the content carries clonezilla_smb_password when: "'STAGED-TEST' not in menu_test_entry.stdout" + - name: "Menu: NOT replaced, so check the live menu's Clonezilla password" + # The Blancco test entry blocked the menu update above: a changed + # clonezilla_smb_password would leave the live menu with the old one + # (Clonezilla back up / restore then cannot mount the share). + shell: | + F="{{ web_root }}/Altiris/iPXE/GetPxeScript.aspx" + grep -q 'ocs_repository=' "$F" || { echo "no Clonezilla repository entries in the live menu"; exit 0; } + if grep -qF "smb3://clonezilla:${CZ_PW}@" "$F"; then echo "password matches"; else + echo "WARNING: the live menu's Clonezilla entries do not carry clonezilla_smb_password;" + echo "remove the Blancco test entry and re-run to update the menu." + fi + environment: + CZ_PW: "{{ clonezilla_smb_password }}" + register: menu_cz_pw + changed_when: false + no_log: true + when: "'STAGED-TEST' in menu_test_entry.stdout" + + - name: "Menu: Clonezilla password check result" + debug: + msg: "{{ menu_cz_pw.stdout }}" + when: "'STAGED-TEST' in menu_test_entry.stdout" + + - name: "Install the Clonezilla back up / restore scripts (fetched by Clonezilla live)" + # GetPxeScript.aspx: ocs_prerun downloads + # http:///clonezilla-scripts/pxe-cz.sh, ocs_live_run runs it. + copy: + src: "{{ usb_mount }}/clonezilla-scripts/" + dest: "{{ web_root }}/clonezilla-scripts/" + mode: '0644' + directory_mode: '0755' + # Boot files (80) and the iPXE menu (4433) are served on the imaging IP # only, never on the GE NIC. 9009 (webapp) stays on all addresses: admins # reach it from the GE side (see pxe-server.conf below). @@ -1118,7 +1174,9 @@ browseable = yes read only = no guest ok = no - valid users = pxe-upload + # clonezilla: the menu's back up / restore entries (this share + # only); pxe-upload: admins and manual Clonezilla runs. + valid users = pxe-upload clonezilla force user = root comment = Clonezilla backup images @@ -1171,6 +1229,20 @@ executable: /bin/bash changed_when: false + - name: "Create the clonezilla Samba user ([clonezilla] share only)" + # System user, no home, no login shell. Only [clonezilla] lists it in + # valid users. smbpasswd runs every time so a changed + # clonezilla_smb_password takes effect (the menu is updated with it). + shell: | + id clonezilla >/dev/null 2>&1 || useradd -r -M -d /nonexistent -s /usr/sbin/nologin clonezilla + printf '%s\n%s\n' "$CZ_PW" "$CZ_PW" | smbpasswd -a -s clonezilla >/dev/null + environment: + CZ_PW: "{{ clonezilla_smb_password }}" + args: + executable: /bin/bash + changed_when: false + no_log: true + - name: "Deploy nf_conntrack TCP timeout sysctl for PXE workload" copy: src: "{{ usb_mount }}/pxe-server-helpers/99-pxe-conntrack.conf" diff --git a/playbook/site-vars.example.yml b/playbook/site-vars.example.yml index 5bfab75..b1d19bd 100644 --- a/playbook/site-vars.example.yml +++ b/playbook/site-vars.example.yml @@ -35,3 +35,10 @@ corp_nic_mac: "" # e.g. "20:88:10:af:0d:e5" # Optional: kernel the Blancco UEFI entries boot (must be installed; see # scripts/download-packages.sh BLANCCO_KVER). # blancco_kver: "7.0.0-34-generic" + +# Recommended: the password of the Samba account "clonezilla", which can +# open only the [clonezilla] share (Clonezilla back up / restore by machine +# number). The iPXE menu carries it, so treat it as an imaging-LAN secret, +# not a real one. 8-64 characters from A-Z a-z 0-9 . _ ~ - (default +# "clonezilla-backup"). +# clonezilla_smb_password: "change-me-per-site" diff --git a/tests/vm/README.md b/tests/vm/README.md index bb709f4..2d2c600 100644 --- a/tests/vm/README.md +++ b/tests/vm/README.md @@ -88,5 +88,61 @@ grow the client while both are off, and put them back afterwards: virsh -c qemu:///system setmem pxetest-client 4096M --config Menu keys from the host: `virsh -c qemu:///system send-key KEY_DOWN` -(separator lines are skipped: Down x1 = Blancco, x3 = Clonezilla, -x4 = Reboot), then KEY_ENTER; startnet answers are KEY_ + KEY_ENTER. +(separator lines are skipped: Down x1 = Blancco, x2 = Blancco debug, +x3 = Clonezilla: back up, x4 = Clonezilla: restore, x5 = Clonezilla manual, +x6 = Clonezilla manual 32-bit, x7 = Reboot), then KEY_ENTER; startnet +answers are KEY_ + KEY_ENTER. + +## Clonezilla back up / restore by machine number (VM test plan) + +Covers GetPxeScript.aspx (:cz_backup, :cz_restore, :cz_auto64/32), +playbook/clonezilla-scripts/pxe-cz.sh, the clonezilla Samba account and the +webapp's Clonezilla backups page. Re-run the playbook first (or rebuild) so +the server has the new menu, /var/www/html/clonezilla-scripts/pxe-cz.sh and +the clonezilla user; `--with-boot-tools` is needed for Clonezilla itself +(clonezilla/ and clonezilla-i686/). + +0. Server checks (ssh pxe@): + - `grep -c 'smb3://clonezilla:' /var/www/html/Altiris/iPXE/GetPxeScript.aspx` = 2, + and no `__CLONEZILLA_SMB_PASSWORD__` left in it. + - `curl -s http://172.16.9.1/clonezilla-scripts/pxe-cz.sh | head -1` = `#!/bin/bash`. + - `pdbedit -L | grep ^clonezilla:`; `smbclient //172.16.9.1/clonezilla -U clonezilla% -c ls` + works, `smbclient //172.16.9.1/winpeapps -U clonezilla% -c ls` is + refused (NT_STATUS_ACCESS_DENIED): the account opens only its share. + - Blancco Releases page: add and remove a test entry still works. +1. Back up as 6501, 64-bit Secure Boot client (pxetest-client, raise RAM to + 3-4 GiB as for Blancco: the live system plus the squashfs need about + 1.5 GiB). Put something on its 1G disk first (e.g. boot client2's WinPE + and `diskpart` a partition, or `qemu-img` a small NTFS/ext4 image). + Menu Down x3, Enter. At `Machine number:` press Enter on empty (asks + again), then type 6501. Expect: Clonezilla boots through the shim (no + Secure Boot error), no language/keyboard questions, no password on + screen, the share mounts, one disk so no disk question, image + 6501-YYYY-MM-DD-HHMM (server time, Eastern) is saved, "Checking the saved + image", SUCCESS banner with serial and model, power off after 60 s. +2. Same as 6501 from a 32-bit BIOS client: a legacy-BIOS VM (SeaBIOS) with + `-cpu` lacking long mode (e.g. `pentium3` + or qemu32) so `cpuid --ext 29` fails and :cz_auto32 boots + clonezilla-i686. Expect pxe-meta.txt bits=32, firmware=bios, + clonezilla=clonezilla-live-3.2.0-5-i686. Also type `65 01` once: + Clonezilla says the number is not valid and asks again. +3. Webapp (Tools > Clonezilla backups): both images under 6501, newest + first, serial and model from the VM's DMI (QEMU: serial empty unless + `` sets one; set `VMSER1`), size, + disks/partitions, Clonezilla version, 64/32-bit, Check = Checked. + Download .tar of one image and `tar tvf` it. +4. Restore to a second, LARGER blank disk: attach a blank 4G disk to the + client (`virsh attach-disk ... --config`, then boot). Menu Down x4, + machine 6501. Expect only 6501's images, newest first; pick one; the + disk list shows both disks with model and size; pick the 4G one; type + YES. Clonezilla runs restoredisk with -k1 -r: afterwards `sfdisk -l` + (Clonezilla shell, s at the end prompt) shows the partitions scaled to + 4G and the filesystem grown. Picking the 1G source disk for a 4G image + (repeat with the disks swapped) must be refused as SMALLER. +5. Mismatch flag: back up the second client (different `` serial) + also as 6501. The page shows "Serial mismatch" on 6501 and marks the odd + serial; Move to another machine... (e.g. 6502) clears it, and restore of + 6502 lists the moved image. +6. Retention: Settings > Keep last 1, Apply retention... lists the older + 6501 image only; Delete removes it, the newest stays. + diff --git a/webapp/app.py b/webapp/app.py index 7b6571d..a32139b 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -44,6 +44,7 @@ from werkzeug.utils import secure_filename import config from services import ( blancco_evidence, + clonezilla_images, hardware_coverage, blancco_report, dell_updates, @@ -551,28 +552,110 @@ def images_rename(image_type): # Routes - Clonezilla Backups # --------------------------------------------------------------------------- -@app.route("/backups") -def clonezilla_backups(): - backups = [] - if os.path.isdir(config.CLONEZILLA_SHARE): - for f in sorted(os.listdir(config.CLONEZILLA_SHARE)): - fpath = os.path.join(config.CLONEZILLA_SHARE, f) - if os.path.isfile(fpath) and f.lower().endswith(".zip"): - stat = os.stat(fpath) - backups.append({ - "filename": f, - "machine": os.path.splitext(f)[0], - "size": stat.st_size, - "modified": stat.st_mtime, - }) +def _backups_page(plan=None, plan_keep=None): + images = clonezilla_images.list_images() + zips = clonezilla_images.list_zips() + groups = clonezilla_images.grouped(images, zips) + st = clonezilla_images.settings() return render_template( "backups.html", - backups=backups, + groups=groups, + backups=zips, + image_count=len(images), + mismatches=sum(1 for g in groups if g["mismatch"]), + space=clonezilla_images.disk_space(st["warn_percent"]), + cz_settings=st, + plan=plan, + plan_keep=plan_keep, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) +@app.route("/backups") +def clonezilla_backups(): + """Clonezilla image directories and machine-number zips on the share, + grouped by machine number (services/clonezilla_images.py).""" + return _backups_page() + + +@app.route("/backups/image//delete", methods=["POST"]) +def clonezilla_image_delete(name): + try: + clonezilla_images.delete_image(name) + audit("BACKUP_IMAGE_DELETE", name) + flash(f"Deleted image {name}.", "success") + except (clonezilla_images.ImageError, OSError) as ex: + flash(str(ex), "danger") + return redirect(url_for("clonezilla_backups")) + + +@app.route("/backups/image//move", methods=["POST"]) +def clonezilla_image_move(name): + try: + new = clonezilla_images.move_image(name, request.form.get("machine", "")) + audit("BACKUP_IMAGE_MOVE", f"{name} -> {new}") + flash(f"Moved {name} to machine {clonezilla_images.norm_machine(request.form.get('machine'))}: now {new}.", "success") + except (clonezilla_images.ImageError, OSError) as ex: + flash(str(ex), "danger") + return redirect(url_for("clonezilla_backups")) + + +@app.route("/backups/image//tar") +def clonezilla_image_tar(name): + """The image directory as a tar, streamed file by file (no temp copy).""" + try: + gen = clonezilla_images.tar_stream(name) + except clonezilla_images.ImageError as ex: + flash(str(ex), "danger") + return redirect(url_for("clonezilla_backups")) + audit("BACKUP_IMAGE_DOWNLOAD", name) + resp = Response(stream_with_context(gen), mimetype="application/x-tar") + resp.headers["Content-Disposition"] = f'attachment; filename="{secure_filename(name) or "image"}.tar"' + resp.headers["X-Accel-Buffering"] = "no" + return resp + + +@app.route("/backups/settings", methods=["POST"]) +def clonezilla_settings(): + try: + st = clonezilla_images.save_settings(request.form.get("keep_last", ""), request.form.get("warn_percent", "")) + audit("BACKUP_SETTINGS", f"keep_last={st['keep_last']} warn_percent={st['warn_percent']}") + flash("Clonezilla backup settings saved." + ("" if not st["keep_last"] else + " Nothing is deleted until you press Apply retention and confirm."), "success") + except clonezilla_images.ImageError as ex: + flash(str(ex), "danger") + return redirect(url_for("clonezilla_backups")) + + +@app.route("/backups/retention") +def clonezilla_retention_preview(): + """What 'keep last N per machine' would delete. Shows only; the apply + is a separate POST of exactly these names.""" + keep = clonezilla_images.settings()["keep_last"] + if not keep: + flash("Set Keep last N images per machine first.", "warning") + return redirect(url_for("clonezilla_backups")) + return _backups_page(plan=clonezilla_images.retention_plan(keep), plan_keep=keep) + + +@app.route("/backups/retention/apply", methods=["POST"]) +def clonezilla_retention_apply(): + keep = clonezilla_images.settings()["keep_last"] + names = request.form.getlist("name") + if not keep or not names: + flash("Nothing to delete.", "warning") + return redirect(url_for("clonezilla_backups")) + deleted, skipped = clonezilla_images.apply_retention(names, keep) + for n in deleted: + audit("BACKUP_RETENTION_DELETE", f"{n} (keep last {keep})") + msg = f"Retention (keep last {keep}): deleted {len(deleted)} image{'s' if len(deleted) != 1 else ''}." + if skipped: + msg += f" Kept {len(skipped)} that are no longer due for deletion: {', '.join(skipped[:5])}{'...' if len(skipped) > 5 else ''}." + flash(msg, "success" if not skipped else "warning") + return redirect(url_for("clonezilla_backups")) + + @app.route("/backups/upload", methods=["POST"]) def clonezilla_upload(): if "backup_file" not in request.files: diff --git a/webapp/config.py b/webapp/config.py index af2b9b3..4040a35 100644 --- a/webapp/config.py +++ b/webapp/config.py @@ -32,6 +32,8 @@ DISPLAY_TZ = os.environ.get("DISPLAY_TZ", "America/New_York") HARDWARE_STATE = os.environ.get("HARDWARE_STATE", "/var/lib/pxe-webapp/hardware-seen.json") BIOS_DIR = os.environ.get("BIOS_DIR", "/srv/samba/winpeapps/_shared/BIOS") BLANCCO_STATE = os.environ.get("BLANCCO_STATE", "/var/lib/pxe-webapp/blancco-state.json") +# Clonezilla backups page settings (keep last N per machine, disk warning). +CLONEZILLA_STATE = os.environ.get("CLONEZILLA_STATE", "/var/lib/pxe-webapp/clonezilla-settings.json") # PXE server address the Blancco preferences must point the report share at # (blancco-init.sh greps for it too); checked when a release is staged. BLANCCO_SERVER = os.environ.get("BLANCCO_SERVER", "172.16.9.1") diff --git a/webapp/services/clonezilla_images.py b/webapp/services/clonezilla_images.py new file mode 100644 index 0000000..cc29f86 --- /dev/null +++ b/webapp/services/clonezilla_images.py @@ -0,0 +1,544 @@ +"""Clonezilla images on the clonezilla share, grouped by machine number. + +The boot menu's "Clonezilla: back up this PC" (playbook/clonezilla-scripts/ +pxe-cz.sh) saves -[-n] image directories with a +pxe-meta.txt (key=value: machine, serial, model, disk, date, Clonezilla +version, bits, save/check result). Images made by hand (manual mode) have +only Clonezilla's own files; their machine number comes from the directory +name and serial/model from Clonezilla's Info-dmi.txt when it is there. + +Everything here is defensive: the share is written by PCs over SMB, so a +half-written, unreadable or odd directory is listed (or skipped) but never +raises out of list_images(). Paths from the browser go through +resolve_image(), which only accepts a direct child of the share that is a +real (not symlinked) Clonezilla image directory. +""" +from __future__ import annotations + +import json +import os +import re +import shutil +import tarfile +import tempfile +import time +from datetime import datetime, timezone + +import config + +MACHINE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]{0,31}$") +# -YYYY-MM-DD-HHMM, optionally -n when two land in one minute. +NAME_RE = re.compile(r"^(?P[A-Za-z0-9][A-Za-z0-9-]*?)-(?P\d{4}-\d{2}-\d{2}-\d{4})(?:-(?P\d{1,3}))?$") +META_FILE = "pxe-meta.txt" +META_MAX = 64 * 1024 +UNASSIGNED = "" # group key for images with no machine number +IN_PROGRESS_SECONDS = 6 * 3600 + +DEFAULT_SETTINGS = {"keep_last": 0, "warn_percent": 85} + + +class ImageError(Exception): + """A refused action (bad name, missing image, clash); the message is shown.""" + + +def share(): + return config.CLONEZILLA_SHARE + + +# --- detection and metadata ------------------------------------------------- + +def _listdir(path): + try: + return os.listdir(path) + except OSError: + return [] + + +def is_image_dir(path): + """Clonezilla image dir: a 'disk' or 'parts' file plus an image file + (partclone *-ptcl-img*, dd *.dd-img*, ntfsclone/partimage *-img.*) or + Clonezilla's clonezilla-img log.""" + if os.path.islink(path) or not os.path.isdir(path): + return False + names = _listdir(path) + if not ("disk" in names or "parts" in names): + return False + for n in names: + if "-ptcl-img" in n or ".dd-img" in n or "-img." in n or n == "clonezilla-img": + return True + return False + + +def _read_text(path, limit=META_MAX): + try: + with open(path, "rb") as f: + return f.read(limit).decode("utf-8", "replace") + except OSError: + return "" + + +def read_meta(path): + """pxe-meta.txt as a dict ({} when missing or unreadable).""" + out = {} + for line in _read_text(os.path.join(path, META_FILE)).splitlines(): + if "=" not in line or line.startswith("#"): + continue + k, v = line.split("=", 1) + k = k.strip() + if k and k not in out: + out[k] = v.strip() + return out + + +def _dmi_fallback(path): + """Serial, model and maker from Clonezilla's Info-dmi.txt (dmidecode + output), System Information block.""" + text = _read_text(os.path.join(path, "Info-dmi.txt"), 512 * 1024) + out = {} + block = False + for line in text.splitlines(): + s = line.strip() + if s == "System Information": + block = True + continue + if block: + if not line.startswith(("\t", " ")) and s: + break + for key, name in (("serial", "Serial Number"), ("model", "Product Name"), ("manufacturer", "Manufacturer")): + if s.startswith(name + ":") and key not in out: + out[key] = s.split(":", 1)[1].strip() + return out + + +def _version_fallback(path): + """'clonezilla-live-3.3.3-37-amd64' from clonezilla-img ("Saved by ...").""" + m = re.search(r"Saved by (clonezilla-live-[A-Za-z0-9._-]+?)\.?\s", _read_text(os.path.join(path, "clonezilla-img"), 16 * 1024) + " ") + return m.group(1) if m else "" + + +def _bits_from_version(v): + v = (v or "").lower() + if "amd64" in v or "x86_64" in v: + return "64" + if "i686" in v or "i386" in v or "i586" in v: + return "32" + return "" + + +def _words(path, name): + return _read_text(os.path.join(path, name), 4096).split() + + +def dir_size(path): + """Bytes on disk, like du (st_blocks), symlinks not followed.""" + total = 0 + for root, dirs, files in os.walk(path, followlinks=False): + for n in files: + try: + st = os.lstat(os.path.join(root, n)) + except OSError: + continue + total += getattr(st, "st_blocks", 0) * 512 or st.st_size + return total + + +def _parse_stamp(stamp): + try: + return datetime.strptime(stamp, "%Y-%m-%d-%H%M") + except (TypeError, ValueError): + return None + + +def norm_machine(m): + return (m or "").strip().upper() + + +def image_info(name, with_size=True): + """One image directory as a dict for the page and the actions.""" + path = os.path.join(share(), name) + meta = read_meta(path) + m = NAME_RE.match(name) + machine = norm_machine(meta.get("machine")) + source = "meta" if machine else "" + if not machine and m: + machine, source = norm_machine(m.group("machine")), "name" + if machine and not MACHINE_RE.match(machine): + machine, source = UNASSIGNED, "" + try: + mtime = os.stat(path).st_mtime + except OSError: + mtime = 0 + when = None + if meta.get("date_utc"): + try: + when = datetime.strptime(meta["date_utc"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc).timestamp() + except ValueError: + when = None + if when is None and m and _parse_stamp(m.group("stamp")): + when = _parse_stamp(m.group("stamp")).timestamp() # server-local naming time + if when is None: + when = mtime + fb = {} if (meta.get("serial") and meta.get("model")) else _dmi_fallback(path) + version = meta.get("clonezilla") or _version_fallback(path) + save = meta.get("save", "") + check = meta.get("check", "") + if save == "failed": + status = "failed" + elif check == "ok": + status = "ok" + elif check == "failed": + status = "check-failed" + elif meta: + status = "unchecked" + elif time.time() - mtime < IN_PROGRESS_SECONDS: + status = "in-progress" + else: + status = "manual" + disk_size = meta.get("disk_size_bytes", "") + return { + "name": name, + "machine": machine, + "machine_source": source, + "when": when, + "mtime": mtime, + "stamp": m.group("stamp") if m else "", + "suffix": (m.group("n") or "") if m else "", + "size": dir_size(path) if with_size else 0, + "disks": _words(path, "disk"), + "parts": _words(path, "parts"), + "serial": meta.get("serial") or fb.get("serial", ""), + "model": meta.get("model") or fb.get("model", ""), + "manufacturer": meta.get("manufacturer") or fb.get("manufacturer", ""), + "clonezilla": version, + "bits": meta.get("bits") or _bits_from_version(version), + "firmware": meta.get("firmware", ""), + "disk_model": meta.get("disk_model", ""), + "disk_size": int(disk_size) if disk_size.isdigit() else 0, + "date_text": meta.get("date", ""), + "check_date": meta.get("check_date", ""), + "save": save, + "check": check, + "status": status, + "has_meta": bool(meta), + "good": status in ("ok", "unchecked", "manual"), + } + + +def list_images(with_size=True): + out = [] + for name in sorted(_listdir(share())): + if name.startswith("."): + continue + try: + if not is_image_dir(os.path.join(share(), name)): + continue + out.append(image_info(name, with_size)) + except Exception: # noqa: BLE001 - an odd dir must never break the page + continue + out.sort(key=lambda i: (i["when"], i["suffix"].zfill(3)), reverse=True) + return out + + +def list_zips(): + out = [] + for name in sorted(_listdir(share())): + p = os.path.join(share(), name) + try: + if name.lower().endswith(".zip") and os.path.isfile(p) and not os.path.islink(p): + st = os.stat(p) + out.append({"filename": name, "machine": os.path.splitext(name)[0], + "size": st.st_size, "modified": st.st_mtime}) + except OSError: + continue + return out + + +def grouped(images=None, zips=None): + """Machines, newest activity first (unassigned last). Each: machine, + images (newest first), zips, serials, mismatch, latest.""" + images = list_images() if images is None else images + zips = list_zips() if zips is None else zips + groups = {} + + def g(key): + return groups.setdefault(key, {"machine": key, "images": [], "zips": [], "latest": 0}) + + for i in images: + grp = g(i["machine"]) + grp["images"].append(i) + grp["latest"] = max(grp["latest"], i["when"] or 0) + for z in zips: + key = norm_machine(z["machine"]) if MACHINE_RE.match(z["machine"] or "") else UNASSIGNED + grp = g(key) + grp["zips"].append(z) + grp["latest"] = max(grp["latest"], z["modified"] or 0) + for grp in groups.values(): + serials = [] + for i in grp["images"]: + s = (i["serial"] or "").strip().upper() + if s and s not in serials: + serials.append(s) + grp["serials"] = serials + # Rows whose serial is not the newest image's: the likely typo. + ref = next(((i["serial"] or "").strip().upper() for i in grp["images"] if (i["serial"] or "").strip()), "") + for i in grp["images"]: + s = (i["serial"] or "").strip().upper() + i["serial_differs"] = bool(s and ref and s != ref) + grp["mismatch"] = len(serials) > 1 + grp["size"] = sum(i["size"] for i in grp["images"]) + sum(z["size"] for z in grp["zips"]) + return sorted(groups.values(), key=lambda x: (x["machine"] == UNASSIGNED, -x["latest"], x["machine"])) + + +# --- path safety -------------------------------------------------------------- + +def resolve_image(name): + """Absolute path of image directory `name` (a direct child of the share), + or ImageError.""" + if (not name or name in (".", "..") or name.startswith(".") or "/" in name or "\\" in name + or "\x00" in name or name != os.path.basename(name) or len(name) > 255): + raise ImageError("Invalid image name.") + root = os.path.realpath(share()) + path = os.path.join(root, name) + if os.path.islink(path): + raise ImageError(f"{name} is a link, not an image directory.") + if os.path.dirname(os.path.realpath(path)) != root: + raise ImageError("Invalid image name.") + if not is_image_dir(path): + raise ImageError(f"Image not found: {name}") + return path + + +# --- actions -------------------------------------------------------------------- + +def delete_image(name): + path = resolve_image(name) + shutil.rmtree(path) + return name + + +def _write_meta(path, updates): + meta_path = os.path.join(path, META_FILE) + lines = _read_text(meta_path).splitlines() + keys = set(updates) + kept = [l for l in lines if l.split("=", 1)[0].strip() not in keys] + kept += [f"{k}={v}" for k, v in updates.items()] + fd, tmp = tempfile.mkstemp(dir=path, prefix=".pxe-meta-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write("\n".join(kept) + "\n") + os.chmod(tmp, 0o644) + os.replace(tmp, meta_path) + except BaseException: + if os.path.exists(tmp): + os.remove(tmp) + raise + + +def move_image(name, new_machine): + """Give image `name` to machine `new_machine`: rename the directory to + - and set machine= in pxe-meta.txt. Returns the new name.""" + path = resolve_image(name) + new_machine = norm_machine(new_machine) + if not MACHINE_RE.match(new_machine): + raise ImageError("Machine number: letters, digits and dash only (up to 32, not starting with a dash).") + info = image_info(name, with_size=False) + stamp = info["stamp"] or datetime.fromtimestamp(info["when"] or time.time()).strftime("%Y-%m-%d-%H%M") + existing = {n.lower() for n in _listdir(share())} + base = f"{new_machine}-{stamp}" + new = base if not info["suffix"] else f"{base}-{info['suffix']}" + n = 2 + # SMB is case-insensitive: a case twin is the same name to Windows. + while new.lower() in existing and new.lower() != name.lower(): + new = f"{base}-{n}" + n += 1 + new_path = os.path.join(os.path.dirname(path), new) + if new != name: + if os.path.lexists(new_path) and new.lower() != name.lower(): + raise ImageError(f"{new} already exists.") + os.rename(path, new_path) + updates = {"machine": new_machine} + if info["machine"] and info["machine"] != new_machine: + updates["moved_from"] = info["machine"] + _write_meta(new_path, updates) + return new + + +# --- tar download (streamed, no temp copy) ------------------------------------- + +CHUNK = 1024 * 1024 + + +def tar_stream(name): + """Generator of a POSIX (pax) tar of image `name`, read file by file in + 1 MiB chunks: nothing is staged. A file that shrinks while it is read is + padded with zeros so the archive stays well-formed.""" + path = resolve_image(name) + + def header(ti): + return ti.tobuf(tarfile.PAX_FORMAT, "utf-8", "surrogateescape") + + def gen(): + written = 0 + for root, dirs, files in os.walk(path, followlinks=False): + dirs.sort() + rel_root = os.path.relpath(root, path) + arc_root = name if rel_root == "." else f"{name}/{rel_root}" + ti = tarfile.TarInfo(arc_root) + ti.type = tarfile.DIRTYPE + ti.mode = 0o755 + try: + ti.mtime = int(os.lstat(root).st_mtime) + except OSError: + ti.mtime = int(time.time()) + buf = header(ti) + written += len(buf) + yield buf + for n in sorted(files): + full = os.path.join(root, n) + try: + st = os.lstat(full) + except OSError: + continue + if not os.path.isfile(full) or os.path.islink(full): + continue # sockets, fifos, links: not part of an image + ti = tarfile.TarInfo(f"{arc_root}/{n}") + ti.size = st.st_size + ti.mtime = int(st.st_mtime) + ti.mode = 0o644 + try: + f = open(full, "rb") + except OSError: + continue + buf = header(ti) + written += len(buf) + yield buf + left = ti.size + with f: + while left > 0: + chunk = f.read(min(CHUNK, left)) + if not chunk: + break + left -= len(chunk) + written += len(chunk) + yield chunk + while left > 0: + pad = min(CHUNK, left) + left -= pad + written += pad + yield b"\0" * pad + rem = ti.size % tarfile.BLOCKSIZE + if rem: + pad = tarfile.BLOCKSIZE - rem + written += pad + yield b"\0" * pad + end = b"\0" * (tarfile.BLOCKSIZE * 2) + written += len(end) + rem = written % tarfile.RECORDSIZE + if rem: + end += b"\0" * (tarfile.RECORDSIZE - rem) + yield end + + return gen() + + +# --- disk space ----------------------------------------------------------------- + +def disk_space(warn_percent=None): + warn = settings()["warn_percent"] if warn_percent is None else warn_percent + try: + du = shutil.disk_usage(share()) + except OSError: + return None + pct = (du.used / du.total * 100) if du.total else 0 + return {"total": du.total, "used": du.used, "free": du.free, "percent": pct, + "warn_percent": warn, "warn": pct >= warn} + + +# --- settings and retention --------------------------------------------------- + +def _settings_path(): + return config.CLONEZILLA_STATE + + +def settings(): + out = dict(DEFAULT_SETTINGS) + try: + with open(_settings_path(), encoding="utf-8") as f: + saved = json.load(f) + if isinstance(saved, dict): + for k in DEFAULT_SETTINGS: + if isinstance(saved.get(k), int) and not isinstance(saved.get(k), bool): + out[k] = saved[k] + except (OSError, ValueError): + pass + return out + + +def save_settings(keep_last, warn_percent): + try: + keep_last = int(keep_last) + warn_percent = int(warn_percent) + except (TypeError, ValueError): + raise ImageError("Settings must be whole numbers.") from None + if not 0 <= keep_last <= 100: + raise ImageError("Keep last N images: 0 (off) to 100.") + if not 50 <= warn_percent <= 99: + raise ImageError("Disk space warning: 50 to 99 percent.") + path = _settings_path() + folder = os.path.dirname(path) or "." + os.makedirs(folder, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=folder, prefix=".tmp-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump({"keep_last": keep_last, "warn_percent": warn_percent}, f, indent=1) + os.chmod(tmp, 0o644) + os.replace(tmp, path) + except BaseException: + if os.path.exists(tmp): + os.remove(tmp) + raise + return settings() + + +def retention_plan(keep_last=None, images=None): + """What 'keep the last N images per machine' would delete, never applied + here. Per machine number (unassigned images are never touched): keep the + N newest GOOD images (saved and not failed the check; manual images + count as good), delete every image OLDER than the oldest one kept. + Newer failed or in-progress images stay, so a backup being written now + is never removed and never pushes a good one out.""" + keep_last = settings()["keep_last"] if keep_last is None else keep_last + if not keep_last or keep_last < 1: + return [] + images = list_images() if images is None else images + plan = [] + for grp in grouped(images, []): + if grp["machine"] == UNASSIGNED: + continue + imgs = grp["images"] # newest first + good = [i for i in imgs if i["good"]] + if len(good) <= keep_last: + continue + oldest_kept = good[keep_last - 1] + cut = imgs.index(oldest_kept) + for i in imgs[cut + 1:]: + plan.append({"machine": grp["machine"], "name": i["name"], "when": i["when"], + "size": i["size"], "status": i["status"]}) + return plan + + +def apply_retention(names, keep_last=None): + """Delete the posted names that are STILL in the plan (a stale preview + cannot delete an image that has become one of the N newest).""" + planned = {p["name"] for p in retention_plan(keep_last)} + deleted, skipped = [], [] + for n in names: + if n in planned: + try: + delete_image(n) + deleted.append(n) + except (ImageError, OSError): + skipped.append(n) + else: + skipped.append(n) + return deleted, skipped diff --git a/webapp/templates/backups.html b/webapp/templates/backups.html index 4aec16e..5469d2c 100644 --- a/webapp/templates/backups.html +++ b/webapp/templates/backups.html @@ -2,75 +2,222 @@ {% import '_ui.html' as ui %} {% block title %}Clonezilla backups - PXE Manager{% endblock %} -{# Clonezilla backups: one line per zip on the share (machine number, file, - size, date). Download is the row's action; Delete is in the menu and opens - the confirm modal. #} +{# Clonezilla backups, grouped by machine number (newest activity first): + image directories saved by the boot menu's "Clonezilla: back up this PC" + (or by hand) and the older machine-number .zip files. One line per + image; its details (disks, Clonezilla version, check) are in the + expandable body; Download is the row's action, Move and Delete are in + the menu and open confirm modals. #} + +{% macro gb(n) -%} +{%- if n >= 1e12 %}{{ "%.2f"|format(n / 1e12) }} TB{% elif n >= 1e9 %}{{ "%.1f"|format(n / 1e9) }} GB{% else %}{{ "%.0f"|format(n / 1e6) }} MB{% endif -%} +{%- endmacro %} + +{% macro status_chip(i) -%} +{%- if i.status == 'ok' %}{{ ui.chip('Checked', 'ok', 'Clonezilla image check passed' ~ (' ' ~ i.check_date if i.check_date else '')) }} +{%- elif i.status == 'check-failed' %}{{ ui.chip('Check failed', 'fail', 'The image check after saving failed: do not rely on this image') }} +{%- elif i.status == 'failed' %}{{ ui.chip('Save failed', 'fail', 'Clonezilla reported an error while saving: partial image') }} +{%- elif i.status == 'unchecked' %}{{ ui.chip('Not checked', 'warn', 'Saved, but the image check did not run') }} +{%- elif i.status == 'in-progress' %}{{ ui.chip('In progress?', 'run', 'No pxe-meta.txt yet and changed in the last 6 hours: probably still being written') }} +{%- else %}{{ ui.chip('Manual', 'neutral', 'Saved without the PXE menu script: no check result recorded') }}{% endif -%} +{%- endmacro %} {% block extra_head %} {% endblock %} {% block content %} -{% call ui.head('Clonezilla backups', 'Tools', (backups | length) ~ ' backup' ~ ('s' if backups | length != 1) ~ ' on the share') %} +{% set nmach = groups | rejectattr('machine', 'equalto', '') | list | length %} +{% call ui.head('Clonezilla backups', 'Tools', image_count ~ ' image' ~ ('s' if image_count != 1) ~ ' and ' ~ (backups | length) ~ ' zip' ~ ('s' if backups | length != 1) ~ ' for ' ~ nmach ~ ' machine' ~ ('s' if nmach != 1) ~ ' on the share') %} +
+ Settings +
+ + + +
Nothing is deleted automatically: Apply retention shows what would go and asks first.
+ +
+
+ {% if cz_settings.keep_last %} + Apply retention... + {% endif %} {% endcall %}
-
-

Machine backups

- {% if backups %} - - {% for b in backups %} -
-
- {{ b.machine }} - {{ b.filename }} - {{ "%.1f"|format(b.size / 1073741824) }} GB - {{ b.modified | timestamp_fmt }} -
-
- Download - {% call ui.more(b.machine) %} - -
- - {% endcall %} + +{% if plan is not none %} +
+

Apply retention: keep the last {{ plan_keep }} image{{ 's' if plan_keep != 1 }} per machine

+ {% if plan %} +
These {{ plan | length }} image{{ 's' if plan | length != 1 }} ({{ gb(plan | sum(attribute='size')) }}) will be DELETED. This cannot be undone.
+

Kept per machine: the {{ plan_keep }} newest images that saved and passed (or skipped) the check. Newer failed or in-progress images are kept too. Images with no machine number are never touched.

+
    + {% for p in plan %}
  • {{ p.name }} machine {{ p.machine }} | {{ p.when | timestamp_fmt }} | {{ gb(p.size) }}
    • {% endfor %} +
+
+ + {% for p in plan %}{% endfor %} +
+ Cancel +
+
+ {% else %} +
Nothing to delete: no machine has more than {{ plan_keep }} good image{{ 's' if plan_keep != 1 }}.
+
Back
+ {% endif %} +
+{% endif %} + +{% if space %} +
+

Share disk space

+
+ Share disk space + {{ gb(space.used) }} used of {{ gb(space.total) }} ({{ "%.0f"|format(space.percent) }}%) + {{ gb(space.free) }} free + {% if space.warn %}{{ ui.chip('Over ' ~ space.warn_percent ~ '%', 'warn') }}{% endif %}
+
+ {% if space.warn %}
The share is {{ "%.0f"|format(space.percent) }}% full (warning at {{ space.warn_percent }}%). Delete old images, or set Keep last N in Settings and Apply retention. A backup that runs out of space fails part-way.
{% endif %} +
+{% endif %} + +{% if mismatches %} +
{{ mismatches }} machine{{ 's have' if mismatches != 1 else ' has' }} images with different serial numbers (marked Serial mismatch below): a typo in the machine number, or the PC was replaced. Move the wrong images to the right machine number.
+{% endif %} + +
+

Backups by machine

+ {% if groups %} + + {% for g in groups %} + {% set label = g.machine or 'No machine number' %} +
+ +

{{ label }}

+ {{ g.images | length }} image{{ 's' if g.images | length != 1 }}{% if g.zips %}, {{ g.zips | length }} zip{{ 's' if g.zips | length != 1 }}{% endif %} | {{ gb(g.size) }} + + {% if g.mismatch %}{{ ui.chip('Serial mismatch', 'warn', 'Serials: ' ~ (g.serials | join(', '))) }}{{ g.serials | join(', ') }}{% endif %} + + + + + {% for i in g.images %} +
+
+ + {{ i.when | timestamp_fmt }} + {{ i.model or i.name }} + {{ i.serial or '-' }} + {{ gb(i.size) }} + {{ status_chip(i) }} + + +
+
Image
{{ i.name }}
+
Machine
{{ i.machine or '-' }} ({{ {'meta': 'from pxe-meta.txt', 'name': 'from the folder name'}.get(i.machine_source, 'unknown: move it to a machine number') }})
+
PC
{{ i.manufacturer }} {{ i.model or '-' }} | serial {{ i.serial or '-' }}{% if i.serial_differs %} {{ ui.chip('Different serial', 'warn', 'Not the serial of the newest image of this machine') }}{% endif %}
+
Size
{{ gb(i.size) }}
+
Disk
{{ i.disks | join(' ') or '-' }}{% if i.disk_model %} | {{ i.disk_model }}{% endif %}{% if i.disk_size %} | {{ gb(i.disk_size) }}{% endif %}
+
Partitions
{{ i.parts | join(' ') or '-' }}
+
Clonezilla
{{ i.clonezilla or '-' }}{% if i.bits %} | {{ i.bits }}-bit{% endif %}{% if i.firmware %} | {{ i.firmware | upper }}{% endif %}
+
Saved
{{ i.date_text or (i.when | timestamp_fmt) }}{% if i.save == 'failed' %} | save failed{% endif %}
+
Check
{{ {'ok': 'passed', 'failed': 'FAILED', 'not-run': 'not run'}.get(i.check, 'no result recorded') }}{% if i.check_date %} ({{ i.check_date }}){% endif %}
+
+
+
+ Download .tar + {% call ui.more(i.name) %} + + +
+ + {% endcall %} +
+
+ {% endfor %} + + {% for b in g.zips %} +
+
+ {{ b.modified | timestamp_fmt }} + {{ b.filename }} + - + {{ gb(b.size) }} + {{ ui.chip('Zip', 'neutral', 'Uploaded .zip backup') }} + +
+
+ Download + {% call ui.more(b.machine) %} + +
+ + {% endcall %} +
+
+ {% endfor %} +
{% endfor %} {% else %} -
No backups found. Upload a Clonezilla backup .zip to get started.
+
No backups yet. Boot a PC from the network and pick Clonezilla: back up this PC, or upload a .zip.
{% endif %}
-

Naming

-

Name backup files with the machine number, for example 6501.zip. The machine column is the file name without .zip.

-

The Samba share \\pxe-server\clonezilla is also on the network for direct Clonezilla save and restore.

+

How backups are named

+

The boot menu's Clonezilla: back up this PC asks for the machine number and saves <machine>-<YYYY-MM-DD-HHMM> (for example 6501-2026-10-15-1403) to \\pxe-server\clonezilla, then checks it. Clonezilla: restore this PC lists only that machine's images, newest first.

+

Images saved by hand (manual mode) are grouped by the start of their folder name; move them to a machine number if they show under No machine number. Uploaded zips are named with the machine number, for example 6501.zip.

@@ -108,7 +255,7 @@
- + + + + + + + {% endblock %} {% block extra_scripts %} @@ -141,5 +339,21 @@ document.getElementById('deleteModal').addEventListener('show.bs.modal', functio document.getElementById('deleteMachine').textContent = machine; document.getElementById('deleteForm').action = "{{ url_for('clonezilla_delete', filename='__F__') }}".replace('__F__', encodeURIComponent(filename)); }); +document.getElementById('imageDeleteModal').addEventListener('show.bs.modal', function (event) { + var btn = event.relatedTarget; + var name = btn.getAttribute('data-name'); + document.getElementById('imageDeleteName').textContent = name; + document.getElementById('imageDeleteMachine').textContent = btn.getAttribute('data-machine'); + document.getElementById('imageDeleteForm').action = "{{ url_for('clonezilla_image_delete', name='__N__') }}".replace('__N__', encodeURIComponent(name)); +}); +document.getElementById('moveModal').addEventListener('show.bs.modal', function (event) { + var btn = event.relatedTarget; + var name = btn.getAttribute('data-name'); + document.getElementById('moveName').textContent = name; + var inp = document.getElementById('moveMachine'); + inp.value = btn.getAttribute('data-machine') || ''; + document.getElementById('moveForm').action = "{{ url_for('clonezilla_image_move', name='__N__') }}".replace('__N__', encodeURIComponent(name)); + setTimeout(function () { inp.focus(); inp.select(); }, 200); +}); {% endblock %} diff --git a/webapp/tests/test_clonezilla_images.py b/webapp/tests/test_clonezilla_images.py new file mode 100644 index 0000000..779b24a --- /dev/null +++ b/webapp/tests/test_clonezilla_images.py @@ -0,0 +1,422 @@ +"""Clonezilla backups page: image listing, grouping, metadata, serial +mismatch, delete/move safety, retention preview/apply, tar download. + +Run from the webapp folder: + PYTHONPATH=.: python -m unittest discover -s tests -v + +Every config path points into a temp folder (env vars before config is +imported, then patched per test), so nothing on the host is touched. +""" +import io +import json +import os +import shutil +import sys +import tarfile +import tempfile +import time +import unittest +from unittest import mock + +_T = tempfile.mkdtemp(prefix="pxe-cz-test-") +for _k, _sub in {"SAMBA_SHARE": "samba", "CLONEZILLA_SHARE": "cz", "BLANCCO_REPORTS": "br", + "ENROLLMENT_SHARE": "enr", "WEB_ROOT": "www", "IMAGING_DIR": "img", "UPDATES_DIR": "upd", + "TFTP_ROOT": "tftp", "BIOS_DIR": "bios", "UPLOAD_DIR": "upl", "SAMBA_LOG_DIR": "smblog"}.items(): + os.environ.setdefault(_k, os.path.join(_T, _sub)) + os.makedirs(os.environ[_k], exist_ok=True) +for _k in ("AUDIT_LOG", "HARDWARE_STATE", "BLANCCO_STATE", "DNSMASQ_LEASES", "APACHE_ACCESS_LOG", + "DNSMASQ_SYSLOG", "NIC_STATUS_FILE", "NIC_CONF", "DNSMASQ_NIC_CONF", "DNSMASQ_CONF", "CLONEZILLA_STATE"): + os.environ.setdefault(_k, os.path.join(_T, _k.lower())) +os.makedirs(os.path.join(os.environ["ENROLLMENT_SHARE"], "ppkgs"), exist_ok=True) + +WEBAPP = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, WEBAPP) +os.chdir(WEBAPP) + +import config # noqa: E402 +import app as A # noqa: E402 +from services import clonezilla_images as CZ # noqa: E402 + +DMI = """This image was saved from this machine with DMI info at 2025-01-01: +# dmidecode 3.5 +Handle 0x0001, DMI type 1, 27 bytes +System Information +\tManufacturer: Dell Inc. +\tProduct Name: OptiPlex 7010 +\tSerial Number: DMI777 +\tUUID: 4c4c4544 +Handle 0x0002, DMI type 2, 15 bytes +Base Board Information +\tSerial Number: BOARD1 +""" + + +def make_image(root, name, meta=None, disk="sda", parts="sda1 sda2", data=b"x" * 3000, dmi=None, saved_by=None): + d = os.path.join(root, name) + os.makedirs(d) + with open(os.path.join(d, "disk"), "w") as f: + f.write(disk + "\n") + with open(os.path.join(d, "parts"), "w") as f: + f.write(parts + "\n") + with open(os.path.join(d, "sda1.ntfs-ptcl-img.zst"), "wb") as f: + f.write(data) + with open(os.path.join(d, "clonezilla-img"), "w") as f: + f.write("This image was saved by Clonezilla at 2025-01-01.\n") + if saved_by: + f.write(f"Saved by {saved_by}.\n") + if dmi: + with open(os.path.join(d, "Info-dmi.txt"), "w") as f: + f.write(dmi) + if meta is not None: + with open(os.path.join(d, "pxe-meta.txt"), "w") as f: + f.write("".join(f"{k}={v}\n" for k, v in meta.items())) + return d + + +def good_meta(machine, serial="SER1", model="OptiPlex 7090", **kw): + m = {"machine": machine, "serial": serial, "model": model, "manufacturer": "Dell Inc.", + "disk": "sda", "disk_model": "WDC SSD", "disk_size_bytes": "256060514304", + "clonezilla": "clonezilla-live-3.3.3-37-amd64", "bits": "64", "firmware": "uefi", + "save": "ok", "check": "ok", "date": "2026-10-15 14:03 EDT"} + m.update(kw) + return m + + +class Base(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="cz-case-") + self.share = os.path.join(self.tmp, "share") + os.makedirs(self.share) + self.state = os.path.join(self.tmp, "state", "clonezilla-settings.json") + p1 = mock.patch.object(config, "CLONEZILLA_SHARE", self.share) + p2 = mock.patch.object(config, "CLONEZILLA_STATE", self.state) + p1.start() + p2.start() + self.addCleanup(p1.stop) + self.addCleanup(p2.stop) + self.addCleanup(shutil.rmtree, self.tmp, True) + self.c = A.app.test_client() + with self.c.session_transaction() as s: + s["_csrf_token"] = "t" + + def post(self, url, data=None): + d = {"_csrf_token": "t"} + d.update(data or {}) + return self.c.post(url, data=d) + + +class TestListing(Base): + def test_detects_only_image_dirs(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + os.makedirs(os.path.join(self.share, "random-folder")) + os.makedirs(os.path.join(self.share, "only-disk")) + open(os.path.join(self.share, "only-disk", "disk"), "w").close() # disk but no image file + open(os.path.join(self.share, "disk"), "w").close() # a stray file at the top + os.makedirs(os.path.join(self.share, ".hidden-2026-01-01-0000")) + names = [i["name"] for i in CZ.list_images()] + self.assertEqual(names, ["6501-2026-10-15-1403"]) + + def test_symlinked_dir_is_not_listed(self): + outside = os.path.join(self.tmp, "outside") + os.makedirs(outside) + make_image(outside, "6501-2026-01-01-0000") + os.symlink(os.path.join(outside, "6501-2026-01-01-0000"), os.path.join(self.share, "6501-2026-01-01-0000")) + self.assertEqual(CZ.list_images(), []) + + def test_meta_is_read(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", check_date="2026-10-15 14:40 EDT")) + i = CZ.list_images()[0] + self.assertEqual((i["machine"], i["machine_source"]), ("6501", "meta")) + self.assertEqual(i["serial"], "SER1") + self.assertEqual(i["model"], "OptiPlex 7090") + self.assertEqual(i["bits"], "64") + self.assertEqual(i["clonezilla"], "clonezilla-live-3.3.3-37-amd64") + self.assertEqual(i["status"], "ok") + self.assertEqual(i["disks"], ["sda"]) + self.assertEqual(i["parts"], ["sda1", "sda2"]) + self.assertEqual(i["disk_size"], 256060514304) + self.assertGreater(i["size"], 0) + + def test_manual_image_falls_back_to_name_dmi_and_clonezilla_img(self): + d = make_image(self.share, "wj-7001-2025-03-04-0915", dmi=DMI, saved_by="clonezilla-live-3.2.0-5-i686") + old = time.time() - 30 * 86400 + os.utime(d, (old, old)) + i = CZ.list_images()[0] + self.assertEqual((i["machine"], i["machine_source"]), ("WJ-7001", "name")) + self.assertEqual((i["serial"], i["model"], i["manufacturer"]), ("DMI777", "OptiPlex 7010", "Dell Inc.")) + self.assertEqual(i["clonezilla"], "clonezilla-live-3.2.0-5-i686") + self.assertEqual(i["bits"], "32") + self.assertEqual(i["status"], "manual") + + def test_fresh_image_without_meta_is_in_progress(self): + make_image(self.share, "6501-2026-10-15-1403") + self.assertEqual(CZ.list_images()[0]["status"], "in-progress") + + def test_failed_statuses(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", save="failed", check="not-run")) + make_image(self.share, "6501-2026-10-14-1403", good_meta("6501", check="failed")) + st = {i["name"]: i["status"] for i in CZ.list_images()} + self.assertEqual(st, {"6501-2026-10-15-1403": "failed", "6501-2026-10-14-1403": "check-failed"}) + + def test_unreadable_and_odd_contents_never_500(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + bad = make_image(self.share, "6502-2026-10-15-1403", good_meta("6502")) + with open(os.path.join(bad, "pxe-meta.txt"), "wb") as f: + f.write(b"\xff\xfe machine=\x00garbage\nno equals sign\n=\n") + os.makedirs(os.path.join(self.share, "locked-2026-01-01-0000")) + os.chmod(os.path.join(self.share, "locked-2026-01-01-0000"), 0) + self.addCleanup(os.chmod, os.path.join(self.share, "locked-2026-01-01-0000"), 0o755) + with open(os.path.join(self.share, "weird \u00e9 name.zip"), "wb") as f: + f.write(b"PK") + r = self.c.get("/backups") + self.assertEqual(r.status_code, 200) + self.assertIn(b"6501-2026-10-15-1403", r.data) + + def test_share_missing_still_renders(self): + shutil.rmtree(self.share) + r = self.c.get("/backups") + self.assertEqual(r.status_code, 200) + self.assertIn(b"No backups yet", r.data) + + +class TestGrouping(Base): + def test_grouped_newest_first_with_zips_and_unassigned(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + make_image(self.share, "6501-2026-09-01-0800", good_meta("6501")) + make_image(self.share, "6501-2026-10-15-1403-2", good_meta("6501")) + make_image(self.share, "6502-2026-08-01-0800", good_meta("6502", serial="S2")) + d = make_image(self.share, "2026-07-01-10-img") # Clonezilla's default autoname + old = time.time() - 30 * 86400 + os.utime(d, (old, old)) + with open(os.path.join(self.share, "6502.zip"), "wb") as f: + f.write(b"PK") + groups = CZ.grouped() + self.assertEqual([g["machine"] for g in groups], ["6502", "6501", ""]) # zip just written: 6502 newest + g6501 = next(g for g in groups if g["machine"] == "6501") + self.assertEqual([i["name"] for i in g6501["images"]], + ["6501-2026-10-15-1403-2", "6501-2026-10-15-1403", "6501-2026-09-01-0800"]) + g6502 = next(g for g in groups if g["machine"] == "6502") + self.assertEqual([z["filename"] for z in g6502["zips"]], ["6502.zip"]) + page = self.c.get("/backups").data + self.assertIn(b"No machine number", page) + self.assertIn(b"6502.zip", page) + + def test_machine_case_is_normalised(self): + make_image(self.share, "wj-1-2026-10-15-1403", good_meta("wj-1")) + make_image(self.share, "WJ-1-2026-10-14-1403") + self.assertEqual([g["machine"] for g in CZ.grouped()], ["WJ-1"]) + + def test_serial_mismatch_flag(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", serial="AAA111")) + make_image(self.share, "6501-2026-10-14-1403", good_meta("6501", serial="aaa111 ")) + make_image(self.share, "6502-2026-10-14-1403", good_meta("6502", serial="X")) + groups = {g["machine"]: g for g in CZ.grouped()} + self.assertFalse(groups["6501"]["mismatch"]) + make_image(self.share, "6501-2026-10-13-1403", good_meta("6501", serial="BBB222")) + groups = {g["machine"]: g for g in CZ.grouped()} + self.assertTrue(groups["6501"]["mismatch"]) + self.assertEqual(groups["6501"]["serials"], ["AAA111", "BBB222"]) + self.assertEqual([i["serial_differs"] for i in groups["6501"]["images"]], [False, False, True]) + self.assertFalse(groups["6502"]["mismatch"]) + page = self.c.get("/backups").data + self.assertIn(b"Serial mismatch", page) + self.assertIn(b"different serial numbers", page) + + +class TestDeleteMove(Base): + def test_delete_image(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + r = self.post("/backups/image/6501-2026-10-15-1403/delete") + self.assertEqual(r.status_code, 302) + self.assertFalse(os.path.exists(os.path.join(self.share, "6501-2026-10-15-1403"))) + + def test_delete_needs_csrf(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + r = self.c.post("/backups/image/6501-2026-10-15-1403/delete") + self.assertEqual(r.status_code, 403) + self.assertTrue(os.path.isdir(os.path.join(self.share, "6501-2026-10-15-1403"))) + + def test_delete_refuses_unsafe_names(self): + outside = os.path.join(self.tmp, "outside") + make_image(outside, "victim") + os.makedirs(os.path.join(self.share, "plain-folder")) + os.symlink(os.path.join(outside, "victim"), os.path.join(self.share, "link")) + for bad in ("..", ".", "...", "%2e%2e", "..%2Foutside%2Fvictim", "link", "plain-folder", "nope", ".hidden"): + with self.subTest(bad=bad): + self.post(f"/backups/image/{bad}/delete") + for bad in ("..", "../outside/victim", "link", "plain-folder", "", "a/b", "a\\b", "x\x00"): + with self.subTest(direct=bad), self.assertRaises(CZ.ImageError): + CZ.delete_image(bad) + self.assertTrue(os.path.isdir(os.path.join(outside, "victim"))) + self.assertTrue(os.path.isdir(os.path.join(self.share, "plain-folder"))) + self.assertTrue(os.path.islink(os.path.join(self.share, "link"))) + + def test_move_renames_and_updates_meta(self): + make_image(self.share, "6510-2026-10-15-1403", good_meta("6510")) + r = self.post("/backups/image/6510-2026-10-15-1403/move", {"machine": "6501"}) + self.assertEqual(r.status_code, 302) + new = os.path.join(self.share, "6501-2026-10-15-1403") + self.assertTrue(os.path.isdir(new)) + meta = CZ.read_meta(new) + self.assertEqual(meta["machine"], "6501") + self.assertEqual(meta["moved_from"], "6510") + self.assertEqual(meta["serial"], "SER1") + self.assertEqual([i["name"] for i in CZ.list_images()], ["6501-2026-10-15-1403"]) + + def test_move_clash_gets_suffix_and_case_twin_counts(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + make_image(self.share, "6510-2026-10-15-1403", good_meta("6510")) + self.assertEqual(CZ.move_image("6510-2026-10-15-1403", "6501"), "6501-2026-10-15-1403-2") + make_image(self.share, "ab-2026-10-15-1403", good_meta("ab")) + make_image(self.share, "x-2026-10-15-1403", good_meta("x")) + self.assertEqual(CZ.move_image("x-2026-10-15-1403", "AB"), "AB-2026-10-15-1403-2") + + def test_move_manual_image_without_stamp(self): + d = make_image(self.share, "2026-07-01-10-img") + os.utime(d, (1700000000, 1700000000)) + new = CZ.move_image("2026-07-01-10-img", "6501") + self.assertRegex(new, r"^6501-\d{4}-\d{2}-\d{2}-\d{4}$") + self.assertEqual(CZ.read_meta(os.path.join(self.share, new))["machine"], "6501") + + def test_move_refuses_bad_machine(self): + make_image(self.share, "6510-2026-10-15-1403", good_meta("6510")) + for bad in ("", "-6501", "65 01", "../x", "a/b", "x" * 33, "65_01"): + with self.subTest(bad=bad), self.assertRaises(CZ.ImageError): + CZ.move_image("6510-2026-10-15-1403", bad) + self.assertTrue(os.path.isdir(os.path.join(self.share, "6510-2026-10-15-1403"))) + + +class TestRetention(Base): + def _three(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + make_image(self.share, "6501-2026-10-01-1403", good_meta("6501")) + make_image(self.share, "6501-2026-09-01-1403", good_meta("6501")) + make_image(self.share, "6502-2026-09-01-1403", good_meta("6502")) + + def test_settings_validate_and_persist(self): + self.assertEqual(CZ.settings(), {"keep_last": 0, "warn_percent": 85}) + r = self.post("/backups/settings", {"keep_last": "2", "warn_percent": "90"}) + self.assertEqual(r.status_code, 302) + with open(self.state) as f: + self.assertEqual(json.load(f), {"keep_last": 2, "warn_percent": 90}) + for bad in (("-1", "85"), ("101", "85"), ("2", "40"), ("x", "85")): + with self.subTest(bad=bad), self.assertRaises(CZ.ImageError): + CZ.save_settings(*bad) + self.assertEqual(CZ.settings()["keep_last"], 2) + + def test_preview_lists_without_deleting(self): + self._three() + CZ.save_settings(1, 85) + r = self.c.get("/backups/retention") + self.assertEqual(r.status_code, 200) + self.assertIn(b"6501-2026-10-01-1403", r.data) + self.assertIn(b"6501-2026-09-01-1403", r.data) + self.assertIn(b"will be DELETED", r.data) + self.assertEqual(len(CZ.list_images()), 4) + + def test_plan_keeps_newest_good_and_newer_failed(self): + make_image(self.share, "6501-2026-10-16-0900", good_meta("6501", save="failed")) # newer, failed: kept + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", check="failed")) # check failed: not good + make_image(self.share, "6501-2026-10-01-1403", good_meta("6501")) # newest good: kept + make_image(self.share, "6501-2026-09-01-1403", good_meta("6501")) # delete + d = make_image(self.share, "2026-07-01-10-img") # unassigned: never + os.utime(d, (1700000000, 1700000000)) + plan = [p["name"] for p in CZ.retention_plan(1)] + self.assertEqual(plan, ["6501-2026-09-01-1403"]) + self.assertEqual(CZ.retention_plan(0), []) + self.assertEqual(CZ.retention_plan(5), []) + + def test_apply_deletes_only_planned_names(self): + self._three() + CZ.save_settings(1, 85) + r = self.c.post("/backups/retention/apply", data={ + "_csrf_token": "t", + "name": ["6501-2026-10-01-1403", "6501-2026-09-01-1403", "6501-2026-10-15-1403", "6502-2026-09-01-1403", "../x"]}) + self.assertEqual(r.status_code, 302) + left = sorted(i["name"] for i in CZ.list_images()) + self.assertEqual(left, ["6501-2026-10-15-1403", "6502-2026-09-01-1403"]) + + def test_stale_preview_cannot_delete_a_now_kept_image(self): + self._three() + CZ.save_settings(1, 85) + planned = [p["name"] for p in CZ.retention_plan()] + shutil.rmtree(os.path.join(self.share, "6501-2026-10-15-1403")) # newest removed meanwhile + deleted, skipped = CZ.apply_retention(planned) + self.assertEqual(deleted, ["6501-2026-09-01-1403"]) + self.assertEqual(skipped, ["6501-2026-10-01-1403"]) + self.assertTrue(os.path.isdir(os.path.join(self.share, "6501-2026-10-01-1403"))) + + def test_apply_needs_csrf_and_setting(self): + self._three() + r = self.c.post("/backups/retention/apply", data={"name": ["6501-2026-09-01-1403"]}) + self.assertEqual(r.status_code, 403) + self.post("/backups/retention/apply", {"name": "6501-2026-09-01-1403"}) # keep_last 0: nothing + self.assertEqual(len(CZ.list_images()), 4) + + +class TestTarAndSpace(Base): + def test_tar_download_streams_the_folder(self): + data = os.urandom(3 * 1024 * 1024 + 123) + d = make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"), data=data) + os.makedirs(os.path.join(d, "sub")) + with open(os.path.join(d, "sub", "f.txt"), "w") as f: + f.write("hello") + os.symlink("/etc/passwd", os.path.join(d, "evil-link")) + r = self.c.get("/backups/image/6501-2026-10-15-1403/tar") + self.assertEqual(r.status_code, 200) + self.assertEqual(r.mimetype, "application/x-tar") + self.assertIn('filename="6501-2026-10-15-1403.tar"', r.headers["Content-Disposition"]) + self.assertTrue(r.is_streamed) + body = r.get_data() + self.assertEqual(len(body) % tarfile.RECORDSIZE, 0) + with tarfile.open(fileobj=io.BytesIO(body)) as t: + names = t.getnames() + self.assertIn("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst", names) + self.assertIn("6501-2026-10-15-1403/sub/f.txt", names) + self.assertNotIn("6501-2026-10-15-1403/evil-link", names) + self.assertEqual(t.extractfile("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst").read(), data) + self.assertEqual(t.extractfile("6501-2026-10-15-1403/sub/f.txt").read(), b"hello") + + def test_tar_of_shrinking_file_stays_valid(self): + d = make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"), data=b"y" * 5000) + gen = CZ.tar_stream("6501-2026-10-15-1403") + chunks = [] + for c in gen: + chunks.append(c) + if len(chunks) == 1: # first header out: shrink a file before it is read + with open(os.path.join(d, "sda1.ntfs-ptcl-img.zst"), "wb") as f: + f.write(b"y" * 10) + with tarfile.open(fileobj=io.BytesIO(b"".join(chunks))) as t: + m = t.getmember("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst") + self.assertEqual(len(t.extractfile(m).read()), m.size) + + def test_tar_bad_name_redirects(self): + for bad in ("..", "nope", "plain"): + with self.subTest(bad=bad): + r = self.c.get(f"/backups/image/{bad}/tar") + self.assertEqual(r.status_code, 302) + + def test_disk_space_and_warning(self): + make_image(self.share, "6501-2026-10-15-1403", good_meta("6501")) + sp = CZ.disk_space(50) + self.assertGreater(sp["total"], 0) + fake = shutil._ntuple_diskusage(100, 95, 5) + with mock.patch.object(CZ.shutil, "disk_usage", return_value=fake): + self.assertTrue(CZ.disk_space(90)["warn"]) + self.assertFalse(CZ.disk_space(96)["warn"]) + page = self.c.get("/backups").data + self.assertIn(b"Share disk space", page) + self.assertIn(b"95% full", page) + + def test_zip_routes_still_work(self): + with open(os.path.join(self.share, "6501.zip"), "wb") as f: + f.write(b"PK-zip") + r = self.c.get("/backups/download/6501.zip") + self.assertEqual(r.data, b"PK-zip") + r.close() + self.post("/backups/delete/6501.zip") + self.assertFalse(os.path.exists(os.path.join(self.share, "6501.zip"))) + + +if __name__ == "__main__": + unittest.main()