mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-02 10:52:22 +00:00
fix(vm): reject failed ISO downloads and repair image sources
This commit is contained in:
martino
@@ -0,0 +1,263 @@
|
||||
"""The VM wizard must not accept failed ISO downloads as installation media.
|
||||
|
||||
Run its real functions against a loopback HTTP fixture and stub only external
|
||||
Proxmox/UI services. No product script is sourced, no guest operation runs,
|
||||
and no public network or real ISO storage is accessed.
|
||||
"""
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
CREATOR = ROOT / "scripts/vm/vm_creator.sh"
|
||||
PAYLOAD = b"fixture installation media\x00\x01\n"
|
||||
REAL_WGET = shutil.which("wget")
|
||||
|
||||
|
||||
class MediaHandler(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
if self.path == "/missing.iso":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
if self.path == "/partial.iso":
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Length", "1024")
|
||||
self.end_headers()
|
||||
self.wfile.write(b"partial")
|
||||
self.close_connection = True
|
||||
return
|
||||
body = b"" if self.path == "/empty.iso" else PAYLOAD
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Content-Disposition", 'attachment; filename="remote-name.iso"')
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, format, *args):
|
||||
pass
|
||||
|
||||
|
||||
class VmIsoDownloadTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if REAL_WGET is None:
|
||||
raise RuntimeError("These offline tests require wget, as the VM wizard does")
|
||||
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), MediaHandler)
|
||||
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
||||
cls.thread.start()
|
||||
cls.http = f"http://127.0.0.1:{cls.server.server_port}"
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.shutdown()
|
||||
cls.server.server_close()
|
||||
cls.thread.join()
|
||||
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory(prefix="vm-iso-download-")
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.directory = Path(self.temp.name)
|
||||
self.iso = self.directory / "installation media.iso"
|
||||
self.qm_log = self.directory / "qm.log"
|
||||
self.wget_log = self.directory / "wget.log"
|
||||
self.bin = self.directory / "bin"
|
||||
self.bin.mkdir()
|
||||
# Route only the optional fixed VirtIO/SourceForge endpoints to our
|
||||
# fixture. Wget still performs the actual HTTP transfer and file I/O.
|
||||
assert REAL_WGET is not None
|
||||
wrapper = self.bin / "wget"
|
||||
wrapper.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
'printf "%s\\n" "$*" >> "$WGET_LOG"\n'
|
||||
'args=("$@")\n'
|
||||
'for i in "${!args[@]}"; do\n'
|
||||
' case "${args[$i]}" in\n'
|
||||
' https://fedorapeople.org/*) args[$i]="$VIRTIO_FIXTURE_URL";;\n'
|
||||
' https://sourceforge.net/*) args[$i]="$HTTP_FIXTURE/success.iso";;\n'
|
||||
' esac\n'
|
||||
' if [[ "${args[$i]}" == http://* || "${args[$i]}" == https://* ]]; then\n'
|
||||
' [[ "${args[$i]}" == "$HTTP_FIXTURE"/* ]] || exit 98\n'
|
||||
' fi\n'
|
||||
'done\n'
|
||||
'for i in "${!args[@]}"; do\n'
|
||||
' if [[ "${args[$i]}" == -O ]]; then\n'
|
||||
' case "${args[$((i+1))]}" in "$FIXTURE_DIR"/*) ;; *) exit 97;; esac\n'
|
||||
' fi\n'
|
||||
'done\n'
|
||||
f"exec {shlex.quote(REAL_WGET)} --no-config --no-proxy --tries=1 --timeout=1 "
|
||||
'"${args[@]}"\n'
|
||||
)
|
||||
wrapper.chmod(0o755)
|
||||
|
||||
def run_wizard(self, route="/success.iso", windows=False, overwrite=True, url=None):
|
||||
source = CREATOR.read_text()
|
||||
functions = "\n\n".join(re.findall(r"^function \w+\(\) \{\n.*?^\}", source, re.M | re.S))
|
||||
self.assertIn("function create_vm()", functions)
|
||||
prelude = r'''
|
||||
translate() { printf '%s' "$1"; }
|
||||
msg_ok() { printf 'OK:%s\n' "$*"; }
|
||||
msg_error() { printf 'ERROR:%s\n' "$*"; }
|
||||
msg_warn() { printf 'WARN:%s\n' "$*"; }
|
||||
msg_info() { :; }
|
||||
msg_success() { :; }
|
||||
qm() { printf '%q ' "$@" >> "$QM_LOG"; printf '\n' >> "$QM_LOG"; }
|
||||
pvesm() { printf 'Name Type Status\nfixture dir active\n'; }
|
||||
select_storage_target() { printf fixture; }
|
||||
# No next menu, start, or guest-facing process is launched.
|
||||
bash() { :; }
|
||||
whiptail() {
|
||||
if [[ " $* " == *' --yesno '* ]]; then
|
||||
return "$OVERWRITE_STATUS"
|
||||
fi
|
||||
if [[ ! -f "$DIALOG_ONCE" ]]; then
|
||||
: > "$DIALOG_ONCE"
|
||||
printf '1' >&2
|
||||
return 0
|
||||
fi
|
||||
return 1
|
||||
}
|
||||
IMPORT_DISKS=(); PASSTHROUGH_DISKS=(); VIRTUAL_DISKS=(); CONTROLLER_NVME_PCIS=()
|
||||
VMID=99999; HN=fixture; CORE_COUNT=1; RAM_SIZE=512; BRG=fixture
|
||||
MAC='02:00:00:00:00:01'; VLAN=''; MTU=''; BIOS_TYPE=''; CPU_TYPE=''
|
||||
MACHINE=''; SERIAL_PORT=''; START_VM=no; WIZARD_ADD_GPU=no
|
||||
TAB=''; GN=''; CL=''
|
||||
create_vm
|
||||
'''
|
||||
env = dict(os.environ, PATH=str(self.bin) + os.pathsep + os.environ["PATH"],
|
||||
ISO_PATH=str(self.iso), ISO_URL=url if url is not None else self.http + route,
|
||||
ISO_VOLID="local:iso/installation media.iso", ISO_DIR=str(self.directory),
|
||||
OS_TYPE="2" if windows else "3",
|
||||
QM_LOG=str(self.qm_log), WGET_LOG=str(self.wget_log),
|
||||
DIALOG_ONCE=str(self.directory / "dialog-once"), FIXTURE_DIR=str(self.directory),
|
||||
OVERWRITE_STATUS="0" if overwrite else "1", HTTP_FIXTURE=self.http,
|
||||
VIRTIO_FIXTURE_URL=self.http + route)
|
||||
return subprocess.run(["bash", "--noprofile", "--norc", "-c", functions + prelude],
|
||||
env=env, stdin=subprocess.DEVNULL, capture_output=True,
|
||||
text=True, timeout=15)
|
||||
|
||||
def qm_calls(self):
|
||||
return self.qm_log.read_text() if self.qm_log.exists() else ""
|
||||
|
||||
def test_http_failure_stops_before_creating_a_guest(self):
|
||||
result = self.run_wizard("/missing.iso")
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.qm_calls(), "")
|
||||
self.assertNotIn("OK:ISO image downloaded", result.stdout)
|
||||
self.assertIn("ERROR:Failed to download ISO image", result.stdout)
|
||||
|
||||
def test_partial_transfer_is_not_published_as_cached_media(self):
|
||||
result = self.run_wizard("/partial.iso")
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.qm_calls(), "")
|
||||
self.assertFalse(self.iso.exists(), "A failed transfer must not become reusable media")
|
||||
|
||||
def test_an_empty_cached_download_is_retried(self):
|
||||
self.iso.write_bytes(b"")
|
||||
result = self.run_wizard()
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.iso.read_bytes(), PAYLOAD)
|
||||
self.assertIn("create ", self.qm_calls())
|
||||
|
||||
def test_virtio_download_uses_the_configured_iso_directory(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
result = self.run_wizard(windows=True)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
driver = self.directory / "virtio-win.iso"
|
||||
self.assertTrue(driver.is_file(), result.stdout + result.stderr)
|
||||
self.assertEqual(driver.read_bytes(), PAYLOAD)
|
||||
self.assertIn("-ide3 ", self.qm_calls())
|
||||
|
||||
def test_failed_virtio_fetch_is_not_selected_for_attachment(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
result = self.run_wizard("/missing.iso", windows=True)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertNotIn("-ide3 ", self.qm_calls())
|
||||
self.assertNotIn("OK:VirtIO driver ISO downloaded successfully.", result.stdout)
|
||||
self.assertIn("ERROR:Failed to download VirtIO driver ISO.", result.stdout)
|
||||
self.assertFalse((self.directory / "virtio-win.iso").exists())
|
||||
|
||||
|
||||
def test_successful_virtio_replacement_preserves_existing_permissions(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
driver = self.directory / "virtio-win.iso"
|
||||
driver.write_bytes(b"previous driver")
|
||||
driver.chmod(0o640)
|
||||
result = self.run_wizard(windows=True)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(driver.read_bytes(), PAYLOAD)
|
||||
self.assertEqual(driver.stat().st_mode & 0o777, 0o640)
|
||||
|
||||
def test_empty_http_success_is_rejected(self):
|
||||
result = self.run_wizard("/empty.iso")
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.qm_calls(), "")
|
||||
self.assertFalse(self.iso.exists())
|
||||
self.assertEqual({p.name for p in self.directory.iterdir()}, {"bin", "wget.log"})
|
||||
|
||||
def test_existing_nonempty_media_is_reused_without_fetching(self):
|
||||
self.iso.write_bytes(b"existing installation media")
|
||||
result = self.run_wizard("/missing.iso")
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.iso.read_bytes(), b"existing installation media")
|
||||
self.assertFalse(self.wget_log.exists())
|
||||
self.assertIn("-ide2 ", self.qm_calls())
|
||||
|
||||
def test_sourceforge_download_keeps_the_selected_destination_name(self):
|
||||
result = self.run_wizard(url="https://sourceforge.net/projects/fixture/iso/download")
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.iso.read_bytes(), PAYLOAD)
|
||||
self.assertFalse((self.directory / "remote-name.iso").exists())
|
||||
self.assertEqual({p.name for p in self.directory.iterdir()},
|
||||
{"bin", "wget.log", "qm.log", self.iso.name})
|
||||
|
||||
def test_failed_virtio_replacement_preserves_previous_media(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
driver = self.directory / "virtio-win.iso"
|
||||
driver.write_bytes(b"previous driver")
|
||||
driver.chmod(0o640)
|
||||
result = self.run_wizard("/partial.iso", windows=True)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(driver.read_bytes(), b"previous driver")
|
||||
self.assertEqual(driver.stat().st_mode & 0o777, 0o640)
|
||||
self.assertNotIn("-ide3 ", self.qm_calls())
|
||||
self.assertNotIn("OK:VirtIO driver ISO downloaded successfully.", result.stdout)
|
||||
|
||||
def test_declining_virtio_replacement_reuses_previous_media(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
driver = self.directory / "virtio-win.iso"
|
||||
driver.write_bytes(b"previous driver")
|
||||
result = self.run_wizard("/missing.iso", windows=True, overwrite=False)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(driver.read_bytes(), b"previous driver")
|
||||
self.assertFalse(self.wget_log.exists())
|
||||
self.assertIn("-ide3 ", self.qm_calls())
|
||||
|
||||
def test_empty_cached_virtio_media_is_retried_not_attached(self):
|
||||
self.iso.write_bytes(PAYLOAD)
|
||||
driver = self.directory / "virtio-win.iso"
|
||||
driver.write_bytes(b"")
|
||||
result = self.run_wizard("/missing.iso", windows=True, overwrite=False)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertTrue(self.wget_log.exists())
|
||||
self.assertNotIn("-ide3 ", self.qm_calls())
|
||||
self.assertIn("ERROR:Failed to download VirtIO driver ISO.", result.stdout)
|
||||
|
||||
def test_a_directory_cannot_be_published_as_media(self):
|
||||
self.iso.mkdir()
|
||||
result = self.run_wizard()
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertEqual(self.qm_calls(), "")
|
||||
self.assertTrue(self.iso.is_dir())
|
||||
self.assertEqual(list(self.iso.iterdir()), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user