diff --git a/gns3server/compute/qemu/qemu_vm.py b/gns3server/compute/qemu/qemu_vm.py index eb2f2eb79..908ce15cb 100644 --- a/gns3server/compute/qemu/qemu_vm.py +++ b/gns3server/compute/qemu/qemu_vm.py @@ -158,6 +158,7 @@ class QemuVM(KernelDatapathMixin, BaseNode): self._hdc_disk_interface = "none" self._hdd_disk_interface = "none" self._cdrom_image = "" + self._cdrom2_image = "" self._bios_image = "" self._boot_priority = "c" self._mac_address = "" @@ -524,33 +525,74 @@ class QemuVM(KernelDatapathMixin, BaseNode): else: self._cdrom_image = "" + @property + def cdrom2_image(self): + return self._cdrom2_image + + @cdrom2_image.setter + def cdrom2_image(self, image): + self._cdrom2_image = self.manager.get_abs_image_path(image, self.working_dir) if image else "" + async def update_property(self, name, value): """ Update Qemu VM properties. """ + previous = getattr(self, name) setattr(self, name, value) - if name == "cdrom_image": - # let the guest know about the new cdrom image - await self._update_cdrom_image() + try: + if name in {"cdrom_image", "cdrom2_image"}: + await self._update_cdrom_image(name) + except (Exception, asyncio.CancelledError): + # Old media may have been deleted while QEMU still has it open. + # Restore the stored path without resolving it again. + setattr(self, f"_{name}", previous) + raise self._command_line_changed = True - async def _update_cdrom_image(self): - """ - Update the cdrom image path for the Qemu guest OS - """ + async def _update_cdrom_image(self, name="cdrom_image"): + """Change media only on an existing drive, validating the affected slot.""" + if not self.is_running(): + return + image = getattr(self, name) + self._cdrom_option(name) + device = "ide1-cd0" if name == "cdrom_image" else "ide1-cd1" + reader, writer = await self._open_qemu_monitor_connection_vm() + if reader is None or writer is None: + raise QemuError("Could not connect to the QEMU monitor to change CD/DVD media") - if self.is_running(): - if self._cdrom_image: - self._cdrom_option() # this will check the cdrom image is accessible - await self._control_vm("eject -f ide1-cd0") - await self._control_vm(f"change ide1-cd0 {self._cdrom_image}") - log.debug( - f'QEMU VM "{self._name}" [{self._id}] has changed the cdrom image path to {self._cdrom_image}' - ) + async def command(text): + writer.write((text + "\n").encode("utf-8")) + await writer.drain() + # Consume the echo first: a filename may itself contain "(qemu) ". + await asyncio.wait_for(reader.readline(), timeout=3) + output = await asyncio.wait_for(reader.readuntil(b"(qemu) "), timeout=3) + response = output.decode("utf-8", errors="replace") + response = response.removesuffix("(qemu) ").strip() + if "Error:" in response or (text != "info block" and response): + raise QemuError(response) + return response + + try: + await asyncio.wait_for(reader.readuntil(b"(qemu) "), timeout=3) + blocks = await command("info block") + if not re.search(rf"(?m)^{device}[: ]", blocks): + raise QemuError("Stop the VM before adding this CD/DVD drive") + if image: + filename = image + if not image.isascii() or any(ord(char) < 32 for char in image): + # HMP's line editor cannot reliably handle non-ASCII bytes. + # A JSON block filename carries them as ASCII escapes, which + # the block layer decodes after the HMP parser has run. + filename = "json:" + json.dumps({"driver": "raw", "file": {"driver": "file", "filename": image}}) + await command(f"change {device} -f {json.dumps(filename)}") else: - await self._control_vm("eject -f ide1-cd0") - log.debug(f'QEMU VM "{self._name}" [{self._id}] has ejected the cdrom image') + await command(f"eject -f {device}") + log.debug('QEMU VM "%s" has changed %s to %s', self._name, name, image) + except (asyncio.TimeoutError, asyncio.IncompleteReadError, OSError) as e: + raise QemuError(f"Could not change CD/DVD media: {e}") from e + finally: + writer.close() @property def bios_image(self): @@ -2387,10 +2429,10 @@ class QemuVM(KernelDatapathMixin, BaseNode): return options - def _cdrom_option(self): + def _cdrom_option(self, name=None): options = [] - if self._cdrom_image: + if name in (None, "cdrom_image") and self._cdrom_image: if not os.path.isfile(self._cdrom_image) or not os.path.exists(self._cdrom_image): if os.path.islink(self._cdrom_image): raise QemuError( @@ -2401,6 +2443,17 @@ class QemuVM(KernelDatapathMixin, BaseNode): if self._hdc_disk_image: raise QemuError("You cannot use a disk image on hdc disk and a CDROM image at the same time") options.extend(["-cdrom", self._cdrom_image.replace(",", ",,")]) + if name in (None, "cdrom2_image") and self._cdrom2_image: + if not os.path.isfile(self._cdrom2_image): + raise QemuError(f"cdrom image '{self._cdrom2_image}' is not accessible") + interface = self._hdd_disk_interface + if self._create_config_disk and interface == "none": + interface = self._hda_disk_interface + if (self._hdd_disk_image or self._create_config_disk) and interface == "ide": + raise QemuError("You cannot use an IDE hdd disk and a second CDROM image at the same time") + options.extend( + ["-drive", f"file={self._cdrom2_image.replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on"] + ) return options def _bios_option(self): @@ -2868,6 +2921,8 @@ class QemuVM(KernelDatapathMixin, BaseNode): answer["cdrom_image"] = self.manager.get_relative_image_path(self._cdrom_image, self.working_dir) answer["cdrom_image_md5sum"] = md5sum(self._cdrom_image, self.working_dir) + answer["cdrom2_image"] = self.manager.get_relative_image_path(self._cdrom2_image, self.working_dir) + answer["cdrom2_image_md5sum"] = md5sum(self._cdrom2_image, self.working_dir) answer["bios_image"] = self.manager.get_relative_image_path(self._bios_image, self.working_dir) answer["bios_image_md5sum"] = md5sum(self._bios_image, self.working_dir) answer["initrd"] = self.manager.get_relative_image_path(self._initrd, self.working_dir) diff --git a/gns3server/controller/export_project.py b/gns3server/controller/export_project.py index f70da2537..0c34d2ed4 100644 --- a/gns3server/controller/export_project.py +++ b/gns3server/controller/export_project.py @@ -208,6 +208,7 @@ async def _patch_project_file( # image files that we need to include in the exported archive images = [] + portable_references = [] try: with open(path) as f: @@ -249,32 +250,54 @@ async def _patch_project_file( if value is None or value.strip() == "": continue - if not keep_compute_ids: # If we keep the original compute we can keep the image path + if not keep_compute_ids: node["properties"][prop] = os.path.basename(value) - if include_images is True: - images.append({"compute_id": compute_id, "image": value, "image_type": node["node_type"]}) + if include_images: + image = {"compute_id": compute_id, "image": value, "image_type": node["node_type"]} + images.append(image) + if not keep_compute_ids: + portable_references.append((node["properties"], prop, image)) if not keep_compute_ids: topology["topology"][ "computes" ] = [] # Strip compute information because could contain secret info like password - local_images = {i["image"] for i in images if i["compute_id"] == "local"} + destinations = _image_export_names(images) + for properties, prop, image in portable_references: + properties[prop] = destinations[(image["compute_id"], image["image_type"], image["image"])] - for image in local_images: - _export_local_image(image, zstream) - - remote_images = {(i["compute_id"], i["image_type"], i["image"]) for i in images if i["compute_id"] != "local"} - - for compute_id, image_type, image in remote_images: - await _export_remote_images(project, compute_id, image_type, image, zstream, temporary_dir) + for (compute_id, image_type, image), export_name in destinations.items(): + if compute_id == "local": + _export_local_image(image, zstream, export_name, image_type) + else: + await _export_remote_images(project, compute_id, image_type, image, zstream, temporary_dir, export_name) zstream.writestr("project.gns3", json.dumps(topology, indent=4, sort_keys=True).encode()) return images -def _export_local_image(image, zstream): +def _image_export_names(images): + """Allocate distinct portable names without changing ordinary image names.""" + destinations = {} + used = set() + for image in images: + key = (image["compute_id"], image["image_type"], image["image"]) + if key in destinations: + continue + filename = os.path.basename(image["image"]) + stem, extension = os.path.splitext(filename) + index = 1 + while (image["image_type"], filename) in used: + index += 1 + filename = f"{stem}-{index}{extension}" + used.add((image["image_type"], filename)) + destinations[key] = filename + return destinations + + +def _export_local_image(image, zstream, export_name=None, image_type=None): """ Exports a local image to the zip file. @@ -285,6 +308,8 @@ def _export_local_image(image, zstream): from ..compute import MODULES for module in MODULES: + if image_type and getattr(module, "_NODE_TYPE", None) != image_type: + continue try: images_directory = module.instance().get_images_directory() except NotImplementedError: @@ -298,13 +323,15 @@ def _export_local_image(image, zstream): path = os.path.join(images_directory, image) if os.path.exists(path): - arcname = os.path.join("images", directory, os.path.basename(image)) + arcname = os.path.join("images", directory, export_name or os.path.basename(image)) _patch_mtime(path) zstream.write(path, arcname) return -async def _export_remote_images(project, compute_id, image_type, image, project_zipfile, temporary_dir): +async def _export_remote_images( + project, compute_id, image_type, image, project_zipfile, temporary_dir, export_name=None +): """ Export specific image from remote compute. """ @@ -334,5 +361,6 @@ async def _export_remote_images(project, compute_id, image_type, image, project_ break await f.write(data) response.close() - arcname = os.path.join("images", image_type, image) + directory = "QEMU" if image_type == "qemu" else image_type + arcname = os.path.join("images", directory, export_name or image) project_zipfile.write(temp_path, arcname=arcname, compress_type=zipfile.ZIP_DEFLATED) diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py index c1036f0ee..5e7d7566c 100644 --- a/gns3server/controller/node.py +++ b/gns3server/controller/node.py @@ -58,6 +58,7 @@ IMAGE_PROPERTIES_BY_NODE_TYPE = { "hdc_disk_image": "qemu", "hdd_disk_image": "qemu", "cdrom_image": "qemu", + "cdrom2_image": "qemu", "initrd": "qemu", "kernel_image": "qemu", "bios_image": "qemu", @@ -900,6 +901,7 @@ class Node: "hdc_disk_image", "hdd_disk_image", "cdrom_image", + "cdrom2_image", "bios_image", "initrd", "kernel_image", diff --git a/gns3server/db/models/images.py b/gns3server/db/models/images.py index 139f9054d..f7b9d96dc 100644 --- a/gns3server/db/models/images.py +++ b/gns3server/db/models/images.py @@ -45,6 +45,23 @@ class Image(BaseTable): last_error = Column(String) templates = relationship("Template", secondary=image_template_map, back_populates="images") + @property + def reference(self): + """An emulator-relative reference, or an absolute additional-root path.""" + import os + + from gns3server.utils import force_unix_path + from gns3server.utils.images import default_images_directory + + root = os.path.abspath(default_images_directory(self.image_type)) + path = os.path.abspath(self.path) + try: + if os.path.commonpath([root, path]) == root: + return force_unix_path(os.path.relpath(path, root)) + except ValueError: # Different drives on Windows. + pass + return force_unix_path(path) + class ImageSyncJob(BaseTable): __tablename__ = "image_sync_jobs" diff --git a/gns3server/db/models/templates.py b/gns3server/db/models/templates.py index 73d18e8b7..ffcaeccef 100644 --- a/gns3server/db/models/templates.py +++ b/gns3server/db/models/templates.py @@ -201,6 +201,7 @@ class QemuTemplate(Template): hdd_disk_image = Column(String) hdd_disk_interface = Column(String) cdrom_image = Column(String) + cdrom2_image = Column(String) initrd = Column(String) kernel_image = Column(String) bios_image = Column(String) diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index bf10fa2f8..1ef935a86 100644 --- a/gns3server/db/tasks.py +++ b/gns3server/db/tasks.py @@ -140,12 +140,16 @@ async def connect_to_db(app: FastAPI) -> None: def upgrade_inventory(connection): from alembic.operations import Operations + from gns3server.db_migrations.versions.a6c8e1b2d490_add_second_qemu_cdrom import ( + upgrade as upgrade_cdrom, + ) from gns3server.db_migrations.versions.d9e8a2b7c401_image_inventory_reconciliation import ( upgrade, ) with Operations.context(MigrationContext.configure(connection)): upgrade() + upgrade_cdrom() await conn.run_sync(upgrade_inventory) await conn.run_sync(run_stamp, alembic_cfg) diff --git a/gns3server/db_migrations/versions/a6c8e1b2d490_add_second_qemu_cdrom.py b/gns3server/db_migrations/versions/a6c8e1b2d490_add_second_qemu_cdrom.py new file mode 100644 index 000000000..262abd8ae --- /dev/null +++ b/gns3server/db_migrations/versions/a6c8e1b2d490_add_second_qemu_cdrom.py @@ -0,0 +1,23 @@ +"""Add a second QEMU CD-ROM image. + +Revision ID: a6c8e1b2d490 +Revises: f7d3a9c2e5b1 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "a6c8e1b2d490" +down_revision = "f7d3a9c2e5b1" +branch_labels = None +depends_on = None + + +def upgrade(): + existing = {column["name"] for column in sa.inspect(op.get_bind()).get_columns("qemu_templates")} + if "cdrom2_image" not in existing: + op.add_column("qemu_templates", sa.Column("cdrom2_image", sa.String())) + + +def downgrade(): + op.drop_column("qemu_templates", "cdrom2_image") diff --git a/gns3server/schemas/compute/qemu_nodes.py b/gns3server/schemas/compute/qemu_nodes.py index 6e26583e2..339f4fdae 100644 --- a/gns3server/schemas/compute/qemu_nodes.py +++ b/gns3server/schemas/compute/qemu_nodes.py @@ -183,8 +183,10 @@ class QemuBase(BaseModel): hdd_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hdd backing file disk image path") hdd_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hdd disk image checksum") hdd_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hdd interface") - cdrom_image: Optional[str] = Field(None, description="QEMU cdrom image path") - cdrom_image_md5sum: Optional[str] = Field(None, description="QEMU cdrom image checksum") + cdrom_image: Optional[str] = Field(None, description="QEMU first cdrom image path") + cdrom2_image: Optional[str] = Field(None, description="QEMU second cdrom image path") + cdrom_image_md5sum: Optional[str] = Field(None, description="QEMU first cdrom image checksum") + cdrom2_image_md5sum: Optional[str] = Field(None, description="QEMU second cdrom image checksum") bios_image: Optional[str] = Field(None, description="QEMU bios image path") bios_image_md5sum: Optional[str] = Field(None, description="QEMU bios image checksum") initrd: Optional[str] = Field(None, description="QEMU initrd path") diff --git a/gns3server/schemas/controller/appliances.py b/gns3server/schemas/controller/appliances.py index ce1a71b98..d3300a675 100644 --- a/gns3server/schemas/controller/appliances.py +++ b/gns3server/schemas/controller/appliances.py @@ -425,7 +425,8 @@ class ApplianceVersionImages(BaseModel): hdb_disk_image: Optional[str] = Field(None, title="Hdc disk image") hdc_disk_image: Optional[str] = Field(None, title="Hdd disk image") hdd_disk_image: Optional[str] = Field(None, title="Hdd diskimage") - cdrom_image: Optional[str] = Field(None, title="cdrom image") + cdrom_image: Optional[str] = Field(None, title="First cdrom image") + cdrom2_image: Optional[str] = Field(None, title="Second cdrom image") class ApplianceVersion(BaseModel): diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py index 17eb73ad3..5d99c296c 100644 --- a/gns3server/schemas/controller/images.py +++ b/gns3server/schemas/controller/images.py @@ -37,6 +37,9 @@ class ImageBase(BaseModel): filename: str = Field(..., description="Image filename") path: str = Field(..., description="Image path") + reference: Optional[str] = Field( + None, description="Image reference relative to its emulator directory, or an absolute additional-root path" + ) image_type: ImageType = Field(..., description="Image type") image_size: int = Field(..., description="Image size in bytes") checksum: str = Field(..., description="Checksum value") diff --git a/gns3server/schemas/controller/templates/qemu_templates.py b/gns3server/schemas/controller/templates/qemu_templates.py index 6f6b29328..399918f55 100644 --- a/gns3server/schemas/controller/templates/qemu_templates.py +++ b/gns3server/schemas/controller/templates/qemu_templates.py @@ -79,7 +79,8 @@ class QemuTemplate(TemplateBase): hdd_disk_interface: Optional[QemuDiskInterfaceType] = Field( QemuDiskInterfaceType.none, description="QEMU hdd interface" ) - cdrom_image: Optional[str] = Field("", description="QEMU cdrom image path") + cdrom_image: Optional[str] = Field("", description="QEMU first cdrom image path") + cdrom2_image: Optional[str] = Field("", description="QEMU second cdrom image path") initrd: Optional[str] = Field("", description="QEMU initrd path") kernel_image: Optional[str] = Field("", description="QEMU kernel image path") bios_image: Optional[str] = Field("", description="QEMU bios image path") diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py index 595c9ec9c..829fe78d2 100644 --- a/gns3server/utils/images.py +++ b/gns3server/utils/images.py @@ -49,6 +49,9 @@ from . import force_unix_path log = logging.getLogger(__name__) +# Include the first ISO 9660 volume descriptor in image type detection. +IMAGE_HEADER_SIZE = 16 * 2048 + 7 + async def list_images(image_type): """ @@ -145,7 +148,7 @@ def inspect_image_file(path, expected_image_type=None, allow_raw_image=False, st info = os.fstat(f.fileno()) if not stat.S_ISREG(info.st_mode) or stat_fingerprint(info) != before: raise ImageChangedError(f"Image changed while opening: {path}") - header = f.read(7) + header = f.read(IMAGE_HEADER_SIZE) if len(header) < 7: raise InvalidImageError(f"Image '{path}' is too small to be valid") image_type = check_valid_image_header(path, header, allow_raw_image) @@ -359,6 +362,8 @@ def check_valid_image_header(path: str, data: bytes, allow_raw_image: bool = Fal elif data[:4] == b"QFI\xfb" or data[:4] == b"KDMV": # for Qemy images: file must be QCOW2 or VMDK return "qemu" + elif data[16 * 2048 + 1 : 16 * 2048 + 6] == b"CD001": + return "qemu" else: if allow_raw_image is True: return "qemu" @@ -380,12 +385,14 @@ async def write_image( if subdirectory is not None: validate_image_upload_name(image_filename) subfolders = validate_image_subdirectory(subdirectory) - # HTTP chunk boundaries need not align with the seven-byte image header. + # HTTP chunk boundaries need not align with the image header or ISO descriptor. iterator = stream.__aiter__() prefix = bytearray() async for chunk in iterator: prefix.extend(chunk) - if len(prefix) >= 7: + # Known disk formats keep their original early detection behavior. + header_size = 7 if prefix.startswith((b"\x7fELF", b"QFI\xfb", b"KDMV")) else IMAGE_HEADER_SIZE + if len(prefix) >= header_size: break if len(prefix) < 7: raise InvalidImageError("The image content is empty or too small to be valid") diff --git a/tests/api/routes/controller/test_images.py b/tests/api/routes/controller/test_images.py index 8e0972b96..adfbcb7ee 100644 --- a/tests/api/routes/controller/test_images.py +++ b/tests/api/routes/controller/test_images.py @@ -476,3 +476,40 @@ class TestImageRoutes: assert len(templates) == 1 assert templates[0].name == "Empty VM" await templates_repo.delete_template(templates[0].template_id) + + +class TestIsoImages: + async def test_iso_uploads_can_be_used_together_in_a_template(self, app, client, config, images_dir): + config.settings.Server.allow_raw_images = False + references = [] + for index, name in enumerate(("installer.ISO", "drivers.iso")): + content = bytes(16 * 2048) + b"\x01CD001\x01" + bytes([index]) * 2041 + response = await client.post( + app.url_path_for("upload_image", image_path=name), + params={"subdirectory": "installers"} if index == 0 else {}, + content=content, + ) + assert response.status_code == status.HTTP_201_CREATED, response.text + assert response.json()["image_type"] == "qemu" + relative = "installers/" + name if index == 0 else name + assert response.json()["path"] == os.path.join(images_dir, "QEMU", relative) + assert response.json()["reference"] == relative + assert Qemu.instance().get_abs_image_path(relative) == response.json()["path"] + references.append(relative) + + response = await client.post( + app.url_path_for("create_template"), + json={ + "name": "Windows installer with drivers", + "template_type": "qemu", + "compute_id": "local", + "cdrom_image": references[0], + "cdrom2_image": references[1], + }, + ) + assert response.status_code == status.HTTP_201_CREATED, response.text + template_id = response.json()["template_id"] + response = await client.get(app.url_path_for("get_template", template_id=template_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json()["cdrom_image"] == references[0] + assert response.json()["cdrom2_image"] == references[1] diff --git a/tests/compute/qemu/test_qemu_vm.py b/tests/compute/qemu/test_qemu_vm.py index 1f4953c27..047a32ad9 100644 --- a/tests/compute/qemu/test_qemu_vm.py +++ b/tests/compute/qemu/test_qemu_vm.py @@ -15,9 +15,11 @@ # along with this program. If not, see . import asyncio +import json import os import stat import sys +from pathlib import Path from unittest import mock from unittest.mock import MagicMock, patch @@ -1039,3 +1041,126 @@ async def test_run_with_kvm_linux_dev_kvm_missing(linux_platform, vm): vm.manager.config.settings.Qemu.require_hardware_acceleration = True with pytest.raises(QemuError): await vm._run_with_hardware_acceleration("qemu-system-x86_64", "") + + +@pytest.mark.asyncio +async def test_two_cdrom_images(vm, images_dir): + installer = Path(images_dir) / "installer.iso" + drivers = Path(images_dir) / "virtio,drivers.iso" + installer.touch() + drivers.touch() + vm.cdrom_image = str(installer) + vm.cdrom2_image = str(drivers) + assert vm._cdrom_option() == [ + "-cdrom", + str(installer), + "-drive", + f"file={str(drivers).replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on", + ] + assert vm.asdict()["cdrom2_image"].endswith("virtio,drivers.iso") + vm.cdrom2_image = "" + assert vm._cdrom_option() == ["-cdrom", str(installer)] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("config_disk", [False, True]) +async def test_second_cdrom_ide_conflict(vm, images_dir, config_disk): + iso = Path(images_dir) / "drivers.iso" + iso.touch() + vm.cdrom2_image = str(iso) + vm._hdd_disk_interface = "ide" + vm._create_config_disk = config_disk + if not config_disk: + vm._hdd_disk_image = str(iso) + with pytest.raises(QemuError, match="second CDROM"): + vm._cdrom_option() + + +@pytest.fixture +def cdrom_monitor(vm): + reader = asyncio.StreamReader() + reader.feed_data(b"QEMU monitor\r\n(qemu) ") + writer = MagicMock() + writer.drain = AsyncioMagicMock() + state = {"blocks": "ide1-cd0: installer.iso\r\nide1-cd1: drivers.iso\r\n", "error": ""} + + def respond(data): + command = data.decode("utf-8").strip() + output = state["blocks"] if command == "info block" else state["error"] + reader.feed_data(data.rstrip(b"\n") + b"\r\n" + output.encode("utf-8") + b"(qemu) ") + + writer.write.side_effect = respond + with ( + patch.object(vm, "is_running", return_value=True), + patch.object( + vm, "_open_qemu_monitor_connection_vm", new_callable=AsyncioMagicMock, return_value=(reader, writer) + ), + ): + yield state, writer + + +@pytest.mark.asyncio +@pytest.mark.parametrize("filename", ["drivers with spaces.iso", "drivers-雪.iso", "drivers (qemu) .iso"]) +async def test_second_cdrom_update(vm, images_dir, cdrom_monitor, filename): + iso = Path(images_dir) / filename + iso.touch() + state, writer = cdrom_monitor + await vm.update_property("cdrom2_image", str(iso)) + assert writer.write.call_args_list[0] == mock.call(b"info block\n") + command = writer.write.call_args_list[1].args[0].decode("ascii") + encoded_filename = json.loads(command.removeprefix("change ide1-cd1 -f ").strip()) + if encoded_filename.startswith("json:"): + assert json.loads(encoded_filename[5:])["file"]["filename"] == str(iso) + else: + assert encoded_filename == str(iso) + assert vm.cdrom2_image == str(iso) + writer.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_second_cdrom_eject(vm, cdrom_monitor): + state, writer = cdrom_monitor + await vm.update_property("cdrom2_image", "") + writer.write.assert_called_with(b"eject -f ide1-cd1\n") + + +@pytest.mark.asyncio +async def test_second_cdrom_hot_add_requires_restart(vm, images_dir, cdrom_monitor): + iso = Path(images_dir) / "drivers.iso" + iso.touch() + state, writer = cdrom_monitor + state["blocks"] = "ide1-cd0: installer.iso\r\n" + vm._command_line_changed = False + with pytest.raises(QemuError, match="Stop the VM"): + await vm.update_property("cdrom2_image", str(iso)) + assert vm.cdrom2_image == "" + assert not vm._command_line_changed + assert writer.write.call_count == 1 + + +@pytest.mark.asyncio +async def test_missing_other_iso_does_not_block_media_swap(vm, images_dir, cdrom_monitor): + iso = Path(images_dir) / "installer.iso" + iso.touch() + vm._cdrom2_image = str(Path(images_dir) / "deleted.iso") + await vm.update_property("cdrom_image", str(iso)) + assert vm.cdrom_image == str(iso) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", ["Error: media change failed", "change: filename expected"]) +@pytest.mark.parametrize("original_deleted", [False, True]) +async def test_cdrom_monitor_error_preserves_previous_image(vm, images_dir, cdrom_monitor, error, original_deleted): + original = Path(images_dir) / "original.iso" + original.touch() + replacement = Path(images_dir) / "replacement.iso" + replacement.touch() + vm.cdrom2_image = str(original) + if original_deleted: + original.unlink() + state, writer = cdrom_monitor + state["error"] = error + "\r\n" + with pytest.raises(QemuError, match=error): + await vm.update_property("cdrom2_image", str(replacement)) + assert vm.cdrom2_image == str(original) + writer.close.assert_called_once() diff --git a/tests/controller/test_appliance_to_template.py b/tests/controller/test_appliance_to_template.py index 9faff12b4..843e52653 100644 --- a/tests/controller/test_appliance_to_template.py +++ b/tests/controller/test_appliance_to_template.py @@ -628,3 +628,22 @@ def test_no_appliance_metadata_when_appliance_has_none(): template = ApplianceToTemplate().new_template(appliance, None, "local") assert "appliance_metadata" not in template + + +@pytest.mark.parametrize("version", [6, 8]) +def test_second_cdrom_survives_appliance_conversion(version): + from copy import deepcopy + + from gns3server.schemas.controller.appliances import ApplianceVersionImages + + slots = {"cdrom_image": "installer.iso", "cdrom2_image": "drivers.iso"} + validated = ApplianceVersionImages.model_validate(slots).model_dump(exclude_none=True) + assert validated == slots + appliance = deepcopy(VYOS_V8) + appliance["versions"][0]["images"] = validated + if version == 6: + appliance["registry_version"] = 6 + appliance["qemu"] = {"arch": "x86_64", "ram": 512, "adapters": 1, "kvm": "allow"} + converted = ApplianceToTemplate().new_template(appliance, appliance["versions"][0], "local") + assert converted["cdrom_image"] == "installer.iso" + assert converted["cdrom2_image"] == "drivers.iso" diff --git a/tests/controller/test_export_project.py b/tests/controller/test_export_project.py index 3f4009c9a..04d8715c0 100644 --- a/tests/controller/test_export_project.py +++ b/tests/controller/test_export_project.py @@ -413,3 +413,49 @@ async def test_export_with_ignoring_snapshots(tmpdir, project): with zipfile.ZipFile(str(tmpdir / "zipfile.zip")) as myzip: assert os.path.join("snapshots", "snap.gns3project") not in [f.filename for f in myzip.filelist] + + +@pytest.mark.asyncio +async def test_export_two_isos_with_identical_filenames(tmpdir, project): + from gns3server.compute.qemu import Qemu + from gns3server.controller.import_project import _import_images + + root = Path(str(tmpdir)) / "QEMU" + installer = root / "windows/install.iso" + drivers = root / "drivers/install.iso" + for image, data in ((installer, b"installer"), (drivers, b"drivers")): + image.parent.mkdir(parents=True) + image.write_bytes(data) + topology = { + "topology": { + "nodes": [ + { + "node_type": "qemu", + "properties": { + "cdrom_image": str(installer), + "cdrom2_image": str(drivers), + }, + } + ] + } + } + Path(project.path, "test.gns3").write_text(json.dumps(topology)) + archive = str(tmpdir / "iso-project.zip") + with aiozipstream.ZipFile() as stream: + with patch.object(Qemu, "get_images_directory", return_value=str(root)): + await export_project(stream, project, str(tmpdir), include_images=True) + await write_file(archive, stream) + with zipfile.ZipFile(archive) as stream: + properties = json.loads(stream.read("project.gns3"))["topology"]["nodes"][0]["properties"] + assert properties["cdrom_image"] != properties["cdrom2_image"] + assert stream.read("images/QEMU/" + properties["cdrom_image"]) == b"installer" + assert stream.read("images/QEMU/" + properties["cdrom2_image"]) == b"drivers" + assert len(stream.namelist()) == len(set(stream.namelist())) + destination = Path(str(tmpdir)) / "unpacked" + stream.extractall(destination) + controller = MagicMock() + imported_images = Path(str(tmpdir)) / "imported-images" + controller.images_path.return_value = str(imported_images) + await _import_images(controller, str(destination / "images")) + assert (imported_images / "QEMU" / properties["cdrom_image"]).read_bytes() == b"installer" + assert (imported_images / "QEMU" / properties["cdrom2_image"]).read_bytes() == b"drivers" diff --git a/tests/utils/test_images.py b/tests/utils/test_images.py index 43568076b..87391d23d 100644 --- a/tests/utils/test_images.py +++ b/tests/utils/test_images.py @@ -187,3 +187,53 @@ async def test_list_images(tmpdir, config): "path": "qemu_image.qcow2", } ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("subdirectory", [None, "installers/linux"]) +async def test_iso_upload_and_discovery(tmp_path, config, subdirectory): + from unittest.mock import AsyncMock + + from gns3server.utils.images import discover_images, inspect_image_file, write_image + + config.settings.Server.images_path = str(tmp_path) + config.settings.Server.additional_images_paths = [] + # ISO 9660 primary volume descriptor; no raw-image fallback is needed. + content = b"\0" * (16 * 2048) + b"\x01CD001\x01" + b"\0" * 2041 + + async def stream(): + # Exercise a descriptor split across upload chunks. + for offset in range(0, len(content), 13): + yield content[offset : offset + 13] + + repo = AsyncMock() + repo.get_image_by_checksum.return_value = None + repo.save_verified_image.side_effect = lambda image: image + result = await write_image( + "installer.ISO", str(tmp_path / "installer.ISO"), stream(), repo, subdirectory=subdirectory + ) + expected = tmp_path / "QEMU" / (subdirectory or "") / "installer.ISO" + assert result["path"] == str(expected) + assert result["image_type"] == "qemu" + assert expected.read_bytes() == content + assert inspect_image_file(expected)["checksum"] == result["checksum"] + assert [image["path"] for image in await discover_images("qemu")] == [str(expected)] + + +def test_iso_extension_alone_is_not_a_valid_image(): + from gns3server.utils.images import InvalidImageError, check_valid_image_header + + with pytest.raises(InvalidImageError): + check_valid_image_header("fake.iso", b"not an ISO image") + + +@pytest.mark.parametrize("images_root", ["images", "QEMU/images"]) +def test_image_reference_uses_configured_root(tmp_path, config, images_root): + from gns3server.db.models.images import Image + + root = tmp_path / images_root + config.settings.Server.images_path = str(root) + image = Image(image_type="qemu", path=str(root / "QEMU/installers/linux.iso")) + assert image.reference == "installers/linux.iso" + image.path = str(tmp_path / "extra/QEMU/linux.iso") + assert image.reference == image.path