Merge pull request #3011 from cristian-ciobanu/feat-#3049

feat(qemu): support two CD/DVD images and consistent ISO handling

gns3server/api/routes/compute/qemu_nodes.py

@@ -135,6 +135,10 @@ async def update_qemu_node(node_data: schemas.QemuUpdate, node: QemuVM = Depends
     """
 
     data = jsonable_encoder(node_data, exclude_unset=True)
+    # Reject interface changes before applying any other property or changing live media.
+    for name in ("cdrom_interface", "cdrom2_interface"):
+        if name in data:
+            node.validate_cdrom_interface(name, data[name])
     # update the console first to avoid issue if updating console type
     node.console = data.pop("console", node.console)
     for name, value in data.items():

gns3server/compute/qemu/qemu_vm.py

@@ -158,6 +158,9 @@ class QemuVM(KernelDatapathMixin, BaseNode):
         self._hdc_disk_interface = "none"
         self._hdd_disk_interface = "none"
         self._cdrom_image = ""
+        self._cdrom2_image = ""
+        self._cdrom_interface = "ide"
+        self._cdrom2_interface = "ide"
         self._bios_image = ""
         self._boot_priority = "c"
         self._mac_address = ""
@@ -524,33 +527,101 @@ 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 ""
+
+    @property
+    def cdrom_interface(self):
+        return self._cdrom_interface
+
+    @cdrom_interface.setter
+    def cdrom_interface(self, interface):
+        self.validate_cdrom_interface("cdrom_interface", interface)
+        self._cdrom_interface = interface
+
+    @property
+    def cdrom2_interface(self):
+        return self._cdrom2_interface
+
+    @cdrom2_interface.setter
+    def cdrom2_interface(self, interface):
+        self.validate_cdrom_interface("cdrom2_interface", interface)
+        self._cdrom2_interface = interface
+
+    def validate_cdrom_interface(self, name, interface):
+        """Validate without mutating the VM, including during API preflight."""
+        if interface not in ("ide", "sata"):
+            raise QemuError("CD/DVD interface must be ide or sata")
+        if interface != getattr(self, name) and self.is_running():
+            raise QemuError("Stop the VM before changing a CD/DVD interface")
+
     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)
+        slot = 0 if name == "cdrom_image" else 1
+        interface = self.cdrom_interface if slot == 0 else self.cdrom2_interface
+        device = f"sata-cdrom{slot}" if interface == "sata" else f"ide1-cd{slot}"
+        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)^{re.escape(device)}(?:\s+\([^\r\n]*\))?:", 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,20 +2458,45 @@ class QemuVM(KernelDatapathMixin, BaseNode):
 
         return options
 
-    def _cdrom_option(self):
-
+    def _cdrom_option(self, name=None):
         options = []
-        if 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(
-                        f"cdrom image '{self._cdrom_image}' linked to '{os.path.realpath(self._cdrom_image)}' is not accessible"
-                    )
-                else:
-                    raise QemuError(f"cdrom image '{self._cdrom_image}' is not accessible")
-            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(",", ",,")])
+        for slot, image_name in enumerate(("cdrom_image", "cdrom2_image")):
+            image = getattr(self, image_name)
+            if name not in (None, image_name) or not image:
+                continue
+            if not os.path.isfile(image):
+                if os.path.islink(image):
+                    raise QemuError(f"cdrom image '{image}' linked to '{os.path.realpath(image)}' is not accessible")
+                raise QemuError(f"cdrom image '{image}' is not accessible")
+            interface = self.cdrom_interface if slot == 0 else self.cdrom2_interface
+            if interface == "sata":
+                # Separate IDs avoid collisions with SATA hard disk controllers.
+                options.extend(
+                    [
+                        "-device",
+                        f"ich9-ahci,id=cdrom-ahci{slot}",
+                        "-drive",
+                        f"file={image.replace(',', ',,')},if=none,id=sata-cdrom{slot},media=cdrom,readonly=on",
+                        "-device",
+                        f"ide-cd,drive=sata-cdrom{slot},bus=cdrom-ahci{slot}.0",
+                    ]
+                )
+            elif slot == 0:
+                if self._hdc_disk_image and self._hdc_disk_interface == "ide":
+                    raise QemuError("You cannot use an IDE hdc disk and a CDROM image at the same time")
+                options.extend(["-cdrom", image.replace(",", ",,")])
+            else:
+                disk_interface = self._hdd_disk_interface
+                if self._create_config_disk and disk_interface == "none":
+                    disk_interface = self._hda_disk_interface
+                if (self._hdd_disk_image or self._create_config_disk) and disk_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={image.replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on",
+                    ]
+                )
         return options
 
     def _bios_option(self):
@@ -2868,6 +2964,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)

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)

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",

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"

gns3server/db/models/templates.py

@@ -201,6 +201,9 @@ class QemuTemplate(Template):
     hdd_disk_image = Column(String)
     hdd_disk_interface = Column(String)
     cdrom_image = Column(String)
+    cdrom2_image = Column(String)
+    cdrom_interface = Column(String, nullable=False, default="ide", server_default="ide")
+    cdrom2_interface = Column(String, nullable=False, default="ide", server_default="ide")
     initrd = Column(String)
     kernel_image = Column(String)
     bios_image = Column(String)

gns3server/db/tasks.py

@@ -140,12 +140,20 @@ 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,
                         )
+                        from gns3server.db_migrations.versions.e2a7c9d4b601_add_cdrom_interfaces import (
+                            upgrade as upgrade_cdrom_interfaces,
+                        )
 
                         with Operations.context(MigrationContext.configure(connection)):
                             upgrade()
+                            upgrade_cdrom()
+                            upgrade_cdrom_interfaces()
 
                     await conn.run_sync(upgrade_inventory)
                     await conn.run_sync(run_stamp, alembic_cfg)

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")

gns3server/db_migrations/versions/e2a7c9d4b601_add_cdrom_interfaces.py

@@ -0,0 +1,22 @@
+"""Add QEMU CD/DVD interfaces."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "e2a7c9d4b601"
+down_revision = "a6c8e1b2d490"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    existing = {column["name"] for column in sa.inspect(op.get_bind()).get_columns("qemu_templates")}
+    for name in ("cdrom_interface", "cdrom2_interface"):
+        if name in existing:
+            continue
+        op.add_column("qemu_templates", sa.Column(name, sa.String(), nullable=False, server_default="ide"))
+
+
+def downgrade():
+    op.drop_column("qemu_templates", "cdrom2_interface")
+    op.drop_column("qemu_templates", "cdrom_interface")

gns3server/schemas/compute/qemu_nodes.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from enum import Enum
-from typing import List, Optional
+from typing import List, Literal, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, Field
@@ -152,6 +152,9 @@ class QemuDiskInterfaceType(str, Enum):
     none = "none"
 
 
+QemuCdromInterface = Literal["ide", "sata"]
+
+
 class QemuBase(BaseModel):
     """
     Common Qemu node properties.
@@ -183,8 +186,12 @@ 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")
+    cdrom_interface: QemuCdromInterface = Field("ide", description="First CD/DVD interface")
+    cdrom2_interface: QemuCdromInterface = Field("ide", description="Second CD/DVD interface")
+    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")

gns3server/schemas/controller/appliances.py

@@ -23,6 +23,7 @@ from uuid import UUID
 from pydantic import AnyUrl, BaseModel, BeforeValidator, Discriminator, EmailStr, Field, Tag
 
 from ..common import ExtraConfig
+from ..compute.qemu_nodes import QemuCdromInterface
 
 # ============================================================================
 # Shared Enums
@@ -398,6 +399,8 @@ class Qemu(BaseModel):
     hdc_disk_interface: Optional[QemuDiskInterface] = Field(
         None, title="Disk interface for the installed hdc_disk_image"
     )
+    cdrom_interface: Optional[QemuCdromInterface] = Field(None, title="First CD/DVD interface")
+    cdrom2_interface: Optional[QemuCdromInterface] = Field(None, title="Second CD/DVD interface")
     hdd_disk_interface: Optional[QemuDiskInterface] = Field(
         None, title="Disk interface for the installed hdd_disk_image"
     )
@@ -425,7 +428,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):
@@ -578,6 +582,8 @@ class QemuPropertiesV8(BaseModel):
     hdc_disk_interface: Optional[QemuDiskInterface] = Field(
         None, title="Disk interface for the installed hdc_disk_image"
     )
+    cdrom_interface: Optional[QemuCdromInterface] = Field(None, title="First CD/DVD interface")
+    cdrom2_interface: Optional[QemuCdromInterface] = Field(None, title="Second CD/DVD interface")
     hdd_disk_interface: Optional[QemuDiskInterface] = Field(
         None, title="Disk interface for the installed hdd_disk_image"
     )

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")

gns3server/schemas/controller/templates/qemu_templates.py

@@ -23,6 +23,7 @@ from gns3server.schemas.compute.qemu_nodes import (
     CustomAdapter,
     QemuAdapterType,
     QemuBootPriority,
+    QemuCdromInterface,
     QemuConsoleType,
     QemuDiskInterfaceType,
     QemuOnCloseAction,
@@ -79,7 +80,10 @@ 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")
+    cdrom_interface: QemuCdromInterface = Field("ide", description="First CD/DVD interface")
+    cdrom2_interface: QemuCdromInterface = Field("ide", description="Second CD/DVD interface")
+    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")

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")

tests/api/routes/compute/test_qemu_nodes.py

@@ -370,6 +370,36 @@ class TestQemuNodesRoutes:
         assert response.json()["hdb_disk_image"] == "linux载.img"
         assert response.json()["ram"] == 1024
 
+    async def test_cdrom_interfaces_update_and_read_back(
+        self, app: FastAPI, compute_client: AsyncClient, qemu_vm: dict
+    ) -> None:
+        url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        response = await compute_client.put(url, json={"cdrom_interface": "sata", "cdrom2_interface": "sata"})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["cdrom_interface"] == "sata"
+        assert response.json()["cdrom2_interface"] == "sata"
+        response = await compute_client.get(url)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["cdrom_interface"] == "sata"
+        assert response.json()["cdrom2_interface"] == "sata"
+
+    @pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
+    async def test_cdrom_interface_rejection_does_not_partially_update(self, app, compute_client, qemu_vm, name):
+        url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        original = (await compute_client.get(url)).json()
+        with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
+            response = await compute_client.put(url, json={"ram": original["ram"] + 128, name: "sata"})
+        assert response.status_code == status.HTTP_409_CONFLICT
+        saved = (await compute_client.get(url)).json()
+        assert saved["ram"] == original["ram"]
+        assert saved[name] == original[name]
+
+    @pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
+    async def test_cdrom_interface_rejects_null(self, app, compute_client, qemu_vm, name):
+        url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        response = await compute_client.put(url, json={name: None})
+        assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
+
     async def test_qemu_nio_create_udp(self, app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
         params = {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"}

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]

tests/api/routes/controller/test_templates.py

@@ -1112,6 +1112,28 @@ class TestDockerTemplate:
 
 
 class TestQemuTemplate:
+    async def test_cdrom_interfaces_persist_on_template_update(self, app, client):
+        params = {
+            "name": "SATA installer",
+            "compute_id": "local",
+            "template_type": "qemu",
+            "cdrom_interface": "sata",
+            "cdrom2_interface": "sata",
+        }
+        response = await client.post(app.url_path_for("create_template"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        template_id = response.json()["template_id"]
+        url = app.url_path_for("get_template", template_id=template_id)
+        saved = (await client.get(url)).json()
+        assert saved["cdrom_interface"] == saved["cdrom2_interface"] == "sata"
+        response = await client.put(
+            app.url_path_for("update_template", template_id=template_id), json={"cdrom_interface": "ide"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        saved = (await client.get(url)).json()
+        assert saved["cdrom_interface"] == "ide"
+        assert saved["cdrom2_interface"] == "sata"
+
     async def test_qemu_template_create(self, app: FastAPI, client: AsyncClient) -> None:
 
         params = {

tests/compute/qemu/test_qemu_vm.py

@@ -15,9 +15,11 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 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,225 @@ 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.mark.asyncio
+async def test_first_cdrom_ide_conflict(vm, images_dir):
+    iso = Path(images_dir) / "installer.iso"
+    iso.touch()
+    vm.cdrom_image = str(iso)
+    vm._hdc_disk_image = str(iso)
+    vm._hdc_disk_interface = "ide"
+    with pytest.raises(QemuError, match="IDE hdc disk"):
+        vm._cdrom_option()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("interface", ["virtio", "scsi", "sata", "nvme"])
+@pytest.mark.parametrize("cdrom_count", [1, 2])
+async def test_four_non_ide_disks_with_cdroms(vm, images_dir, interface, cdrom_count):
+    iso = Path(images_dir) / "installer.iso"
+    iso.touch()
+    vm.cdrom_image = str(iso)
+    for drive in "abcd":
+        disk = Path(images_dir) / f"hd{drive}.qcow2"
+        disk.touch()
+        setattr(vm, f"_hd{drive}_disk_image", str(disk))
+        setattr(vm, f"_hd{drive}_disk_interface", interface)
+    expected = ["-cdrom", str(iso)]
+    if cdrom_count == 2:
+        vm.cdrom2_image = str(iso)
+        expected.extend(["-drive", f"file={iso},if=ide,index=3,media=cdrom,readonly=on"])
+    assert vm._cdrom_option() == expected
+    vm.linked_clone = False
+    with patch.object(vm, "_qemu_img_exec", new_callable=AsyncioMagicMock, return_value=0):
+        disk_options = await vm._disk_options()
+    assert sum(option.startswith("file=") for option in disk_options) == 4
+
+
+@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()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("slot", [0, 1])
+async def test_sata_cdrom_with_ide_disks(vm, images_dir, slot):
+    iso = Path(images_dir) / "installer,drivers.iso"
+    iso.touch()
+    image_name = "cdrom_image" if slot == 0 else "cdrom2_image"
+    interface_name = "cdrom_interface" if slot == 0 else "cdrom2_interface"
+    setattr(vm, image_name, str(iso))
+    setattr(vm, interface_name, "sata")
+    vm._hdc_disk_image = vm._hdd_disk_image = str(iso)
+    vm._hdc_disk_interface = vm._hdd_disk_interface = "ide"
+    assert vm._cdrom_option() == [
+        "-device",
+        f"ich9-ahci,id=cdrom-ahci{slot}",
+        "-drive",
+        f"file={str(iso).replace(',', ',,')},if=none,id=sata-cdrom{slot},media=cdrom,readonly=on",
+        "-device",
+        f"ide-cd,drive=sata-cdrom{slot},bus=cdrom-ahci{slot}.0",
+    ]
+    assert vm.asdict()[interface_name] == "sata"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
+async def test_cdrom_interface_change_requires_stopped_vm(vm, name):
+    with patch.object(vm, "is_running", return_value=True):
+        with pytest.raises(QemuError, match="Stop the VM"):
+            await vm.update_property(name, "sata")
+    assert getattr(vm, name) == "ide"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("slot", [0, 1])
+async def test_sata_cdrom_media_update(vm, images_dir, cdrom_monitor, slot):
+    state, writer = cdrom_monitor
+    setattr(vm, "_cdrom_interface" if slot == 0 else "_cdrom2_interface", "sata")
+    state["blocks"] = f"sata-cdrom{slot} (#block123): old.iso\r\n"
+    iso = Path(images_dir) / "new.iso"
+    iso.touch()
+    await vm.update_property("cdrom_image" if slot == 0 else "cdrom2_image", str(iso))
+    writer.write.assert_called_with(f'change sata-cdrom{slot} -f "{iso}"\n'.encode())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("name", ["cdrom_image", "cdrom2_image"])
+async def test_cdrom_broken_symlink_error_includes_target(vm, images_dir, name):
+    target = Path(images_dir) / "missing.iso"
+    link = Path(images_dir) / "linked.iso"
+    link.symlink_to(target)
+    setattr(vm, f"_{name}", str(link))
+    with pytest.raises(QemuError, match="linked to") as error:
+        vm._cdrom_option()
+    assert str(target) in str(error.value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("slot", [0, 1])
+async def test_sata_cdrom_eject(vm, cdrom_monitor, slot):
+    state, writer = cdrom_monitor
+    setattr(vm, "_cdrom_interface" if slot == 0 else "_cdrom2_interface", "sata")
+    state["blocks"] = f"sata-cdrom{slot}: empty\r\n"
+    await vm.update_property("cdrom_image" if slot == 0 else "cdrom2_image", "")
+    writer.write.assert_called_with(f"eject -f sata-cdrom{slot}\n".encode())

tests/controller/test_appliance_to_template.py

@@ -628,3 +628,52 @@ 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"
+
+
+@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
+def test_v8_cdrom_interface_survives_validation_and_conversion(name):
+    import copy
+
+    appliance = copy.deepcopy(VYOS_V8)
+    appliance["settings"][0]["template_properties"][name] = "sata"
+    validated = ApplianceModel.model_validate(appliance).model_dump(mode="json", exclude_none=True)
+    template = ApplianceToTemplate().new_template(validated, validated["versions"][1], "local")
+    assert template[name] == "sata"
+
+
+def test_legacy_cdrom_interfaces_are_qemu_settings():
+    from gns3server.schemas.controller.appliances import Qemu
+
+    settings = Qemu.model_validate(
+        {
+            "adapter_type": "e1000",
+            "adapters": 1,
+            "ram": 512,
+            "arch": "x86_64",
+            "console_type": "telnet",
+            "kvm": "allow",
+            "cdrom_interface": "sata",
+            "cdrom2_interface": "sata",
+        }
+    )
+    assert settings.model_dump()["cdrom_interface"] == "sata"
+    assert settings.model_dump()["cdrom2_interface"] == "sata"

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"

tests/controller/test_qemu_cdrom_migration.py

@@ -0,0 +1,56 @@
+import pytest
+import sqlalchemy as sa
+from alembic import command, config
+from alembic.operations import Operations
+from alembic.runtime.migration import MigrationContext
+from fastapi import FastAPI
+
+from gns3server.db.models import Base
+from gns3server.db.tasks import connect_to_db
+from gns3server.db_migrations.versions import e2a7c9d4b601_add_cdrom_interfaces as migration
+
+
+def test_cdrom_migration_defaults_idempotence_and_downgrade():
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        connection.exec_driver_sql("CREATE TABLE qemu_templates (id INTEGER PRIMARY KEY)")
+        connection.exec_driver_sql("INSERT INTO qemu_templates (id) VALUES (1)")
+        with Operations.context(MigrationContext.configure(connection)):
+            migration.upgrade()
+            migration.upgrade()
+            assert connection.exec_driver_sql("SELECT cdrom_interface, cdrom2_interface FROM qemu_templates").one() == (
+                "ide",
+                "ide",
+            )
+            migration.downgrade()
+        assert [column["name"] for column in sa.inspect(connection).get_columns("qemu_templates")] == ["id"]
+    engine.dispose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("versioned", [False, True])
+async def test_database_startup_adds_cdrom_interfaces(tmp_path, monkeypatch, versioned):
+    path = tmp_path / "controller.db"
+    engine = sa.create_engine(f"sqlite:///{path}")
+    with engine.connect() as connection:
+        Base.metadata.create_all(connection)
+        connection.exec_driver_sql("ALTER TABLE qemu_templates DROP COLUMN cdrom_interface")
+        connection.exec_driver_sql("ALTER TABLE qemu_templates DROP COLUMN cdrom2_interface")
+        if versioned:
+            cfg = config.Config()
+            cfg.set_main_option("script_location", "gns3server:db_migrations")
+            cfg.attributes["connection"] = connection
+            command.stamp(cfg, migration.down_revision)
+        connection.commit()
+    engine.dispose()
+    monkeypatch.setenv("GNS3_DATABASE_URI", f"sqlite+aiosqlite:///{path}")
+    app = FastAPI()
+    await connect_to_db(app)
+    try:
+        async with app.state._db_engine.connect() as connection:
+            columns = await connection.run_sync(lambda conn: sa.inspect(conn).get_columns("qemu_templates"))
+            assert {"cdrom_interface", "cdrom2_interface"} <= {column["name"] for column in columns}
+            revision = await connection.scalar(sa.text("SELECT version_num FROM alembic_version"))
+            assert revision == migration.revision
+    finally:
+        await app.state._db_engine.dispose()

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