diff --git a/gns3server/api/routes/controller/images.py b/gns3server/api/routes/controller/images.py
index 981774419..c7f6bf3f5 100644
--- a/gns3server/api/routes/controller/images.py
+++ b/gns3server/api/routes/controller/images.py
@@ -44,8 +44,21 @@ from gns3server.controller.controller_error import (
from gns3server.db.repositories.images import ImagesRepository
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.db.repositories.templates import TemplatesRepository
+from gns3server.schemas.controller.images import (
+ ImageCompatibility,
+ ImageCompatibilityCatalog,
+ ImageCompatibilityRequest,
+)
from gns3server.services.image_reconciliation import get_image_reconciliation_service
-from gns3server.utils.image_inventory import ImageLockBusy, contained_path, fingerprint, image_lock, publish_image
+from gns3server.utils.image_inventory import (
+ ImageLockBusy,
+ contained_path,
+ fingerprint,
+ image_lock,
+ publish_image,
+ validate_image_subdirectory,
+ validate_image_upload_name,
+)
from gns3server.utils.images import (
InvalidImageError,
default_images_directory,
@@ -71,6 +84,29 @@ def image_destination(image_path):
return full_path
+@router.get(
+ "/compatibility/catalog",
+ response_model=ImageCompatibilityCatalog,
+ dependencies=[Depends(has_privilege("Image.Allocate"))],
+)
+async def image_compatibility_catalog() -> dict:
+ """List known catalog image sizes so clients can avoid hashing obvious nonmatches."""
+ return Controller.instance().appliance_manager.image_compatibility_catalog()
+
+
+@router.post(
+ "/compatibility",
+ response_model=list[ImageCompatibility],
+ dependencies=[Depends(has_privilege("Image.Allocate"))],
+)
+async def check_image_compatibility(
+ request: ImageCompatibilityRequest,
+ images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
+) -> list[dict]:
+ """Check catalog matches for locally computed MD5 checksums without uploading files."""
+ return await Controller.instance().appliance_manager.check_image_compatibility(request.checksums, images_repo)
+
+
@router.post(
"/sync",
response_model=schemas.ImageSyncJob,
@@ -171,7 +207,8 @@ async def get_images(
@router.post(
"/upload/{image_path:path}",
- response_model=schemas.Image,
+ response_model=schemas.ImageUpload,
+ response_model_exclude_unset=True,
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(has_privilege("Image.Allocate"))],
)
@@ -183,7 +220,12 @@ async def upload_image(
current_user: schemas.User = Depends(get_current_active_user),
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
install_appliances: Optional[bool] = False,
-) -> models.Image:
+ subdirectory: Optional[str] = Query(
+ None,
+ max_length=512,
+ description="Relative subfolder below the detected image type directory; use a plain filename",
+ ),
+) -> models.Image | schemas.ImageUpload:
"""
Upload an image.
@@ -194,13 +236,19 @@ async def upload_image(
"""
image_path = urllib.parse.unquote(image_path)
- image_dir, image_name = os.path.split(image_path)
+ if subdirectory is not None:
+ try:
+ validate_image_subdirectory(subdirectory)
+ validate_image_upload_name(image_path)
+ except ValueError as e:
+ raise ControllerBadRequestError(str(e)) from e
+ image_dir = os.path.dirname(image_path)
# check if the path is within the default images directory
full_path = image_destination(image_path)
# If the client sends X-MD5-Checksum, check for a duplicate before consuming the upload stream
checksum_header = request.headers.get("X-MD5-Checksum")
- if checksum_header:
+ if checksum_header and subdirectory is None:
check_dir = os.path.dirname(full_path) if image_dir else None
duplicate = await images_repo.get_image_by_checksum(checksum_header, check_dir)
if duplicate:
@@ -209,7 +257,14 @@ async def upload_image(
try:
allow_raw_image = Config.instance().settings.Server.allow_raw_images
- image = await write_image(image_path, full_path, request.stream(), images_repo, allow_raw_image=allow_raw_image)
+ image = await write_image(
+ image_path,
+ full_path,
+ request.stream(),
+ images_repo,
+ allow_raw_image=allow_raw_image,
+ subdirectory=subdirectory,
+ )
except (OSError, InvalidImageError, ClientDisconnect, SQLAlchemyError) as e:
service = getattr(request.app.state, "image_reconciliation", None)
if service:
@@ -217,16 +272,30 @@ async def upload_image(
raise ControllerError(f"Could not save image '{image_path}': {e}")
if install_appliances:
- # attempt to automatically create templates based on image checksum
- await Controller.instance().appliance_manager.install_appliances_from_image(
- image_path,
- image.checksum,
- images_repo,
- templates_repo,
- rbac_repo,
- current_user,
- os.path.dirname(image.path),
- )
+ # Snapshot before template operations can roll back and expire ORM attributes.
+ uploaded_image = schemas.Image.model_validate(image).model_dump()
+ # Matching requires the checksum computed from the uploaded contents.
+ try:
+ results = await Controller.instance().appliance_manager.install_appliances_from_image(
+ image_path,
+ image.checksum,
+ images_repo,
+ templates_repo,
+ rbac_repo,
+ current_user,
+ os.path.dirname(image.path),
+ )
+ except (ControllerError, InvalidImageError, OSError, SQLAlchemyError) as e:
+ results = [{"status": "skipped", "reason": f"Template creation failed: {e}"}]
+ if not results:
+ results = [
+ {
+ "status": "skipped",
+ "reason": "No compatible appliance definition found in the server's catalog for this image. "
+ "Manually create a template to use this image.",
+ }
+ ]
+ return schemas.ImageUpload.model_validate({**uploaded_image, "template_results": results})
return image
diff --git a/gns3server/controller/appliance_manager.py b/gns3server/controller/appliance_manager.py
index fc6836116..f4ecdded5 100644
--- a/gns3server/controller/appliance_manager.py
+++ b/gns3server/controller/appliance_manager.py
@@ -16,6 +16,7 @@
# along with this program. If not, see .
import asyncio
+import copy
import json
import logging
import os
@@ -28,6 +29,7 @@ from pydantic import ValidationError
from sqlalchemy.exc import SQLAlchemyError
from gns3server import schemas
+from gns3server.db.models import Image
from gns3server.db.repositories.images import ImagesRepository
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.db.repositories.templates import TemplatesRepository
@@ -38,7 +40,7 @@ from gns3server.utils.images import default_images_directory
from ..config import Config
from ..utils.asyncio import locking
from ..utils.http_client import HTTPClient
-from ..utils.image_inventory import image_lock
+from ..utils.image_inventory import contained_path, image_lock
from ..utils.images import InvalidImageError, read_image_info, write_image
from .appliance import Appliance
from .appliance_to_template import ApplianceToTemplate
@@ -133,9 +135,74 @@ class ApplianceManager:
appliances.append((appliance, image.get("version")))
return appliances
+ def image_compatibility_catalog(self) -> dict:
+ """Return a conservative size filter; uncertain metadata requires full hashing."""
+ sizes = set()
+ unknown = False
+ for appliance in self._appliances.values():
+ for image in appliance.images or []:
+ if not image.get("md5sum"):
+ continue
+ size = image.get("filesize")
+ if isinstance(size, int) and not isinstance(size, bool) and size >= 0:
+ sizes.add(size)
+ else:
+ unknown = True
+ return {"image_sizes": sorted(sizes), "has_unknown_sizes": unknown}
+
+ async def check_image_compatibility(self, checksums: list[str], images_repo: ImagesRepository) -> list[dict]:
+ """Read catalog eligibility without uploading, downloading, or creating templates.
+
+ Other images in the selected batch count as prospective dependencies.
+ Final installation still verifies the uploaded contents and dependencies.
+ """
+ selected = {checksum.lower() for checksum in checksums}
+ available = {}
+ results = []
+ for checksum in dict.fromkeys(checksum.lower() for checksum in checksums):
+ matches = []
+ seen = set()
+ for appliance, image_version in self._find_appliances_from_image_checksum(checksum):
+ try:
+ ApplianceModel.model_validate(appliance.asdict())
+ except ValidationError:
+ continue
+ for version in appliance.versions or []:
+ name = version.get("name")
+ if name != image_version or (appliance.id, name) in seen:
+ continue
+ seen.add((appliance.id, name))
+ missing, downloadable = [], []
+ definitions = {image.get("filename"): image for image in appliance.images or []}
+ for filename in dict.fromkeys((version.get("images") or {}).values()):
+ definition = definitions.get(filename, {})
+ dependency_checksum = (definition.get("md5sum") or "").lower()
+ if dependency_checksum in selected:
+ continue
+ if dependency_checksum not in available:
+ available[dependency_checksum] = bool(
+ dependency_checksum and await images_repo.get_image_by_checksum(dependency_checksum)
+ )
+ if available[dependency_checksum]:
+ continue
+ if definition.get("direct_download_url"):
+ downloadable.append(filename)
+ else:
+ missing.append(filename)
+ matches.append(
+ {
+ "name": appliance.name,
+ "version": name,
+ "missing_images": missing,
+ "downloadable_images": downloadable,
+ }
+ )
+ results.append({"checksum": checksum, "matches": matches})
+ return results
+
async def _download_image(
self, image_dir: str, image_name: str, image_type: str, image_url: str, images_repo: ImagesRepository
- ) -> None:
+ ) -> Image:
"""
Download an image.
"""
@@ -146,8 +213,9 @@ class ApplianceManager:
async with HTTPClient.get(image_url) as response:
if response.status != 200:
raise ControllerError(f"Could not download '{image_name}' due to HTTP error code {response.status}")
- await write_image(
- image_name, image_path, response.content.iter_any(), images_repo, allow_raw_image=True
+ # Explicit paths keep dependencies in the appliance's selected directory.
+ return await write_image(
+ image_path, image_path, response.content.iter_any(), images_repo, allow_raw_image=True
)
except (OSError, InvalidImageError) as e:
raise ControllerError(f"Could not save {image_type} image '{image_path}': {e}")
@@ -169,9 +237,11 @@ class ApplianceManager:
for image in appliance.images or []:
if appliance_file == image.get("filename"):
image_checksum = image.get("md5sum")
- image_in_db = await images_repo.get_image_by_checksum(image_checksum)
+ image_in_db = await images_repo.get_image_by_checksum(image_checksum, image_dir)
+ if image_in_db is None:
+ image_in_db = await images_repo.get_image_by_checksum(image_checksum)
if image_in_db:
- version_images[appliance_key] = image_in_db.filename
+ version_images[appliance_key] = self._image_reference(image_in_db)
else:
# check if the image is on disk but it not yet in the database
image_path = os.path.join(image_dir, appliance_file)
@@ -183,19 +253,30 @@ class ApplianceManager:
f"Image '{image_path}' does not match the appliance checksum"
)
try:
- await images_repo.save_verified_image(image_info)
+ image_in_db = await images_repo.save_verified_image(image_info)
+ if image_in_db is None:
+ raise ControllerError(f"Could not register image '{image_path}'")
+ version_images[appliance_key] = self._image_reference(image_in_db)
except SQLAlchemyError as e:
raise ControllerError(f"Could not register image '{image_path}': {e}") from e
else:
# download the image if there is a direct download URL
direct_download_url = image.get("direct_download_url")
if direct_download_url:
- await self._download_image(
+ image_in_db = await self._download_image(
image_dir, appliance_file, appliance.type, direct_download_url, images_repo
)
+ version_images[appliance_key] = self._image_reference(image_in_db)
else:
raise ControllerError(f"Could not find '{appliance_file}'")
+ @staticmethod
+ def _image_reference(image: Image) -> str:
+ directory = default_images_directory(image.image_type)
+ if contained_path(image.path, directory):
+ return os.path.relpath(image.path, directory).replace(os.sep, "/")
+ return image.path
+
async def _create_template(self, template_data, templates_repo, rbac_repo, current_user) -> dict:
"""
Create a new template and return it as a dict.
@@ -266,6 +347,7 @@ class ApplianceManager:
for version in appliance.versions:
if version.get("name") == image_version:
try:
+ version = copy.deepcopy(version)
await self._find_appliance_version_images(appliance, version, images_repo, image_dir)
template_data = await self._appliance_to_template(appliance, version)
name = template_data.get("name")
@@ -337,6 +419,7 @@ class ApplianceManager:
for appliance_version_info in appliance.versions:
if appliance_version_info.get("name") == version:
try:
+ appliance_version_info = copy.deepcopy(appliance_version_info)
template_type = ApplianceToTemplate().get_template_type(
appliance.asdict(), appliance_version_info
)
diff --git a/gns3server/db/repositories/templates.py b/gns3server/db/repositories/templates.py
index cd3dec706..eb89ce595 100644
--- a/gns3server/db/repositories/templates.py
+++ b/gns3server/db/repositories/templates.py
@@ -29,7 +29,8 @@ from sqlalchemy.orm.session import make_transient
import gns3server.db.models as models
from gns3server.controller.controller_error import ControllerNotFoundError
-from gns3server.utils.image_inventory import image_lock
+from gns3server.utils.image_inventory import image_lock, normalized_path
+from gns3server.utils.images import default_images_directory
from .base import BaseRepository
@@ -137,30 +138,56 @@ class TemplatesRepository(BaseRepository):
await self._db_session.refresh(db_template)
return db_template
- async def get_image(self, image_path: str, *, include_unavailable: bool = False) -> Optional[models.Image]:
+ async def get_image(
+ self,
+ image_path: str,
+ *,
+ include_unavailable: bool = False,
+ image_type: Optional[str] = None,
+ template_id: Optional[UUID] = None,
+ ) -> Optional[models.Image]:
"""
- Get an image by its path.
+ Get an image by its path, preferring its type root for relative references.
+
+ Without image_type, retain legacy filename/suffix matching. Absolute
+ references always match the supplied path regardless of image_type.
+ template_id limits lookup to existing associations when removing an image.
"""
- image_dir, image_name = os.path.split(image_path)
+ lookup_path = os.path.normpath(image_path) if image_type and not os.path.isabs(image_path) else image_path
+ image_dir, image_name = os.path.split(lookup_path)
if os.path.isabs(image_path):
query = select(models.Image).where(models.Image.path == image_path)
elif image_dir:
query = select(models.Image).where(
- models.Image.filename == image_name, models.Image.path.endswith(os.sep + image_path, autoescape=True)
+ models.Image.filename == image_name, models.Image.path.endswith(os.sep + lookup_path, autoescape=True)
)
else:
query = select(models.Image).where(models.Image.filename == image_name)
+ if template_id is not None:
+ query = query.where(models.Image.templates.any(models.Template.template_id == template_id))
if not include_unavailable:
query = query.where(models.Image.availability.in_(["unknown", "available"]))
query = query.order_by(models.Image.image_id)
result = await self._db_session.execute(query)
- images = result.scalars().all()
+ images = list(result.scalars().all())
+ if image_type and not os.path.isabs(image_path):
+ # Type-relative paths must resolve below that type's root first.
+ # Preserve legacy suffix/basename lookup for additional image roots.
+ typed_images = [image for image in images if image.image_type == image_type]
+ if typed_images:
+ expected = normalized_path(os.path.join(default_images_directory(image_type), image_path))
+ for image in typed_images:
+ if normalized_path(image.path) == expected:
+ return image
+ images = typed_images
if len(images) > 1:
log.warning(
- f"Multiple DB entries found for image '{image_path}' "
- f"({len(images)} rows). This indicates a data integrity issue. "
- f"Using the entry with the lowest image_id ({images[0].image_id})."
+ "Multiple images match reference '%s' (%s rows); using '%s' with the lowest image_id (%s)",
+ image_path,
+ len(images),
+ images[0].path,
+ images[0].image_id,
)
return images[0] if images else None
diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py
index cdc1d67ce..a5eb8406c 100644
--- a/gns3server/schemas/__init__.py
+++ b/gns3server/schemas/__init__.py
@@ -29,7 +29,7 @@ from .controller.computes import (
)
from .controller.drawings import Drawing
from .controller.gns3vm import GNS3VM
-from .controller.images import Image, ImageSyncJob, ImageSyncRequest, ImageType
+from .controller.images import Image, ImageSyncJob, ImageSyncRequest, ImageType, ImageUpload
# Controller schemas
from .controller.links import (
@@ -233,6 +233,7 @@ __all__ = [
"ImageSyncJob",
"ImageSyncRequest",
"ImageType",
+ "ImageUpload",
"LLMModelConfigCreate",
"LLMModelConfigData",
"LLMModelConfigInheritedResponse",
diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py
index ed1e98ec9..92ba88775 100644
--- a/gns3server/schemas/controller/images.py
+++ b/gns3server/schemas/controller/images.py
@@ -16,7 +16,7 @@
from datetime import datetime
from enum import Enum
-from typing import Literal, Optional
+from typing import Annotated, Literal, Optional
from pydantic import BaseModel, ConfigDict, Field
@@ -50,6 +50,19 @@ class Image(DateTimeModelMixin, ImageBase):
model_config = ConfigDict(from_attributes=True)
+class ImageTemplateResult(BaseModel):
+ status: Literal["created", "skipped"]
+ name: Optional[str] = None
+ reason: Optional[str] = None
+ template_id: Optional[str] = None
+ version: Optional[str] = None
+ template_type: Optional[str] = None
+
+
+class ImageUpload(Image):
+ template_results: Optional[list[ImageTemplateResult]] = None
+
+
class ImageSyncRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
dry_run: bool = False
@@ -65,3 +78,25 @@ class ImageSyncJob(DateTimeModelMixin):
finished_at: Optional[datetime] = None
counts: dict[str, int]
errors: list[dict[str, str]]
+
+
+class ImageCompatibilityRequest(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+ checksums: list[Annotated[str, Field(pattern=r"^[a-fA-F0-9]{32}$")]] = Field(min_length=1, max_length=1000)
+
+
+class ImageApplianceMatch(BaseModel):
+ name: str
+ version: str
+ missing_images: list[str] = Field(default_factory=list)
+ downloadable_images: list[str] = Field(default_factory=list)
+
+
+class ImageCompatibility(BaseModel):
+ checksum: str
+ matches: list[ImageApplianceMatch]
+
+
+class ImageCompatibilityCatalog(BaseModel):
+ image_sizes: list[int]
+ has_unknown_sizes: bool
diff --git a/gns3server/services/templates.py b/gns3server/services/templates.py
index fc5377ff2..42101eeae 100644
--- a/gns3server/services/templates.py
+++ b/gns3server/services/templates.py
@@ -189,9 +189,9 @@ class TemplatesService:
templates.append(jsonable_encoder(builtin_template))
return templates
- async def _find_image(self, image_path: str):
+ async def _find_image(self, image_path: str, image_type: str):
- image = await self._templates_repo.get_image(image_path)
+ image = await self._templates_repo.get_image(image_path, image_type=image_type)
if not image:
raise ControllerNotFoundError(f"Image '{image_path}' could not be found in the controller database")
if not os.path.exists(image.path):
@@ -203,7 +203,7 @@ class TemplatesService:
images_to_add_to_template = []
if template_type == "dynamips":
if settings.get("image"):
- image = await self._find_image(settings["image"])
+ image = await self._find_image(settings["image"], "ios")
if image.image_type != "ios":
raise ControllerBadRequestError(
f"Image '{image.filename}' type is not 'ios' but '{image.image_type}'"
@@ -211,7 +211,7 @@ class TemplatesService:
images_to_add_to_template.append(image)
elif template_type == "iou":
if settings.get("path"):
- image = await self._find_image(settings["path"])
+ image = await self._find_image(settings["path"], "iou")
if image.image_type != "iou":
raise ControllerBadRequestError(
f"Image '{image.filename}' type is not 'iou' but '{image.image_type}'"
@@ -220,7 +220,7 @@ class TemplatesService:
elif template_type == "qemu":
for key, value in settings.items():
if key.endswith("_image") and value:
- image = await self._find_image(value)
+ image = await self._find_image(value, "qemu")
if image.image_type != "qemu":
raise ControllerBadRequestError(
f"Image '{image.filename}' type is not 'qemu' but '{image.image_type}'"
@@ -274,12 +274,14 @@ class TemplatesService:
raise ControllerNotFoundError(f"Template '{template_id}' not found")
return template
- async def _remove_image(self, template_id: UUID, image_path: str) -> None:
+ async def _remove_image(self, template_id: UUID, image_path: str, image_type: str) -> None:
if not image_path:
return
# Removing an association must also find missing/invalid image rows.
- image = await self._templates_repo.get_image(image_path, include_unavailable=True)
+ image = await self._templates_repo.get_image(
+ image_path, include_unavailable=True, image_type=image_type, template_id=template_id
+ )
if image is None:
return
await self._templates_repo.remove_image_from_template(template_id, image)
@@ -314,13 +316,13 @@ class TemplatesService:
images_to_add_to_template = await self._find_images(db_template.template_type, template_settings)
if isinstance(db_template, models.DynamipsTemplate) and "image" in template_settings:
- await self._remove_image(db_template.template_id, db_template.image)
+ await self._remove_image(db_template.template_id, db_template.image, "ios")
elif isinstance(db_template, models.IOUTemplate) and "path" in template_settings:
- await self._remove_image(db_template.template_id, db_template.path)
+ await self._remove_image(db_template.template_id, db_template.path, "iou")
elif db_template.template_type == "qemu":
for key in template_update.model_dump().keys():
if key.endswith("_image") and key in template_settings:
- await self._remove_image(db_template.template_id, db_template.__dict__[key])
+ await self._remove_image(db_template.template_id, db_template.__dict__[key], "qemu")
db_template = await self._templates_repo.update_template(db_template, template_settings)
for image in images_to_add_to_template:
diff --git a/gns3server/utils/image_inventory.py b/gns3server/utils/image_inventory.py
index 78943ba11..bd2c12112 100644
--- a/gns3server/utils/image_inventory.py
+++ b/gns3server/utils/image_inventory.py
@@ -3,7 +3,9 @@
import asyncio
import hashlib
import os
+import re
import stat
+import uuid
from gns3server.config import Config
@@ -97,3 +99,144 @@ def publish_image(temporary, destination):
"""
os.link(temporary, destination)
os.unlink(temporary)
+
+
+def validate_image_subdirectory(value):
+ """A portable relative folder below the server-selected image type root."""
+ if not value:
+ return []
+ parts = value.split("/")
+ if len(value) > 512 or len(parts) > 8:
+ raise ValueError("Image subfolder is too long or has more than eight levels")
+ for part in parts:
+ if (
+ not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9 ._-]{0,63}", part)
+ or part.endswith((".", " ", ".tmp", ".md5sum"))
+ or part.lower() in ("lib", "lib64")
+ or re.fullmatch(r"(?i)(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?", part)
+ ):
+ raise ValueError(
+ "Use relative subfolders with letters, numbers, spaces, dots, hyphens or underscores; "
+ "hidden, reserved and traversal names are not allowed"
+ )
+ return parts
+
+
+class ImageUploadDirectory:
+ """Keep upload operations anchored to the authorized directory.
+
+ POSIX operations use directory descriptors and never follow descendant
+ symlinks. The portable fallback rejects symlinks/junctions and rechecks the
+ directory before each operation. The configured root is administrator-owned.
+ """
+
+ def __init__(self, root, path):
+ self.root = os.path.realpath(os.path.expanduser(root))
+ self.path = os.path.abspath(path)
+ if contained_path(self.path, normalized_path(root)):
+ self.path = os.path.join(self.root, os.path.relpath(self.path, normalized_path(root)))
+ self.fd = None
+ self.identity = None
+ self.anchored = (
+ all(operation in os.supports_dir_fd for operation in (os.open, os.mkdir, os.link, os.unlink))
+ and hasattr(os, "O_NOFOLLOW")
+ and hasattr(os, "O_DIRECTORY")
+ )
+
+ def __enter__(self):
+ if not contained_path(self.path, self.root):
+ raise OSError("Image destination is outside the configured image directory")
+ os.makedirs(self.root, exist_ok=True)
+ parts = os.path.relpath(self.path, self.root).split(os.sep)
+ if parts == ["."]:
+ parts = []
+ try:
+ current = self.root
+ if self.anchored:
+ self.fd = os.open(current, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+ for part in parts:
+ if self.anchored:
+ try:
+ os.mkdir(part, mode=0o755, dir_fd=self.fd)
+ except FileExistsError:
+ pass
+ child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.fd)
+ os.close(self.fd)
+ self.fd = child
+ else:
+ self._check_components(current)
+ current = os.path.join(current, part)
+ try:
+ os.mkdir(current, mode=0o755)
+ except FileExistsError:
+ pass
+ self._check_components(current)
+ info = os.fstat(self.fd) if self.anchored else os.stat(self.path, follow_symlinks=False)
+ self.identity = (info.st_dev, info.st_ino)
+ self.verify()
+ return self
+ except BaseException:
+ self.__exit__()
+ raise
+
+ def _check_components(self, path):
+ current = self.root
+ relative = os.path.relpath(path, self.root)
+ for part in [] if relative == "." else relative.split(os.sep):
+ current = os.path.join(current, part)
+ info = os.lstat(current)
+ reparse = getattr(info, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+ if stat.S_ISLNK(info.st_mode) or reparse:
+ raise OSError("Image upload folders must not be symlinks or junctions")
+
+ def verify(self):
+ self._check_components(self.path)
+ info = os.stat(self.path, follow_symlinks=False)
+ if (info.st_dev, info.st_ino) != self.identity or not contained_path(os.path.realpath(self.path), self.root):
+ raise OSError("Image upload directory changed during upload")
+
+ def temporary(self):
+ self.verify()
+ name = f".gns3-upload-{uuid.uuid4().hex}.tmp"
+ flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
+ if self.anchored:
+ fd = os.open(name, flags | os.O_NOFOLLOW, 0o700, dir_fd=self.fd)
+ else:
+ fd = os.open(os.path.join(self.path, name), flags, 0o700)
+ return fd, name
+
+ def publish(self, temporary, filename):
+ self.verify()
+ if self.anchored:
+ os.link(temporary, filename, src_dir_fd=self.fd, dst_dir_fd=self.fd, follow_symlinks=False)
+ else:
+ os.link(os.path.join(self.path, temporary), os.path.join(self.path, filename))
+ self.remove(temporary)
+ self.verify()
+ return fingerprint(os.path.join(self.path, filename))
+
+ def remove(self, name):
+ if self.anchored:
+ os.unlink(name, dir_fd=self.fd)
+ else:
+ self.verify()
+ os.unlink(os.path.join(self.path, name))
+
+ def __exit__(self, *args):
+ if self.fd is not None:
+ os.close(self.fd)
+ self.fd = None
+
+
+def validate_image_upload_name(name):
+ if (
+ not name
+ or len(name) > 255
+ or name.startswith(".")
+ or name.endswith((".", " ", ".tmp", ".md5sum"))
+ or any(ord(c) < 32 or ord(c) == 127 or c in '/\\:%<>"|?*' for c in name)
+ or re.fullmatch(r"(?i)(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?", name)
+ ):
+ raise ValueError(
+ "Subfolder uploads require a plain, non-hidden image filename without path separators or reserved characters"
+ )
diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py
index 4db2c2f8d..595c9ec9c 100644
--- a/gns3server/utils/images.py
+++ b/gns3server/utils/images.py
@@ -18,11 +18,18 @@ import asyncio
import hashlib
import os
import stat
-import tempfile
import aiofiles
-from gns3server.utils.image_inventory import contained_path, fingerprint, image_lock, publish_image, stat_fingerprint
+from gns3server.utils.image_inventory import (
+ ImageUploadDirectory,
+ contained_path,
+ fingerprint,
+ image_lock,
+ stat_fingerprint,
+ validate_image_subdirectory,
+ validate_image_upload_name,
+)
try:
import importlib_resources
@@ -365,9 +372,14 @@ async def write_image(
images_repo: ImagesRepository,
check_image_header=True,
allow_raw_image=False,
+ subdirectory=None,
) -> models.Image:
image_dir, image_name = os.path.split(image_filename)
+ subfolders = []
+ 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.
iterator = stream.__aiter__()
prefix = bytearray()
@@ -379,55 +391,68 @@ async def write_image(
raise InvalidImageError("The image content is empty or too small to be valid")
image_type = check_valid_image_header(image_path, bytes(prefix), allow_raw_image or not check_image_header)
if not image_dir:
- image_path = os.path.abspath(os.path.join(default_images_directory(image_type), image_name))
+ image_path = os.path.abspath(os.path.join(default_images_directory(image_type), *subfolders, image_name))
root = os.path.realpath(os.path.expanduser(Config.instance().settings.Server.images_path))
if not contained_path(os.path.realpath(image_path), root):
raise InvalidImageError(f"Image destination is outside the configured image directory: {image_path}")
- os.makedirs(os.path.dirname(image_path), exist_ok=True)
- descriptor, tmp_path = tempfile.mkstemp(prefix=".gns3-upload-", suffix=".tmp", dir=os.path.dirname(image_path))
- os.close(descriptor)
- checksum = hashlib.md5()
- try:
- async with aiofiles.open(tmp_path, "wb") as f:
- await f.write(prefix)
- checksum.update(prefix)
- async for chunk in iterator:
- await f.write(chunk)
- checksum.update(chunk)
- image_size = os.path.getsize(tmp_path)
- async with image_lock(image_path):
- if os.path.lexists(image_path):
- raise InvalidImageError(
- f"File '{image_path}' already exists, please choose a different name or remove the existing image"
- )
- checksum_str: str = checksum.hexdigest()
- duplicate_image = await images_repo.get_image_by_checksum(checksum_str, os.path.dirname(image_path))
- if duplicate_image:
- raise InvalidImageError(
- f"Image '{duplicate_image.filename}' with the same checksum "
- f"already exists in '{os.path.dirname(image_path)}'"
- )
- os.chmod(tmp_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
- publish_image(tmp_path, image_path)
- # Complete files survive a database failure so the next scan can
- # recover them. Never compensate by unlinking a published image.
- image = await images_repo.save_verified_image(
- dict(
- image_name=image_name,
- image_type=image_type,
- image_size=image_size,
- path=image_path,
- checksum=checksum_str,
- checksum_algorithm="md5",
- file_fingerprint=fingerprint(image_path),
- )
- )
- if image is None:
- raise InvalidImageError(f"Failed to save image '{image_name}' to database")
- return image
- finally:
+ root = Config.instance().settings.Server.images_path
+ with ImageUploadDirectory(root, os.path.dirname(image_path)) as directory:
+ descriptor, temporary = directory.temporary()
+ checksum = hashlib.md5()
try:
- if os.path.exists(tmp_path):
- os.remove(tmp_path)
- except OSError:
- log.warning(f"Could not remove '{tmp_path}'")
+ # Own the descriptor explicitly so cancellation cannot leak it.
+ async with aiofiles.open(descriptor, "wb", closefd=False) as f:
+ await f.write(prefix)
+ checksum.update(prefix)
+ async for chunk in iterator:
+ await f.write(chunk)
+ checksum.update(chunk)
+ image_size = os.fstat(descriptor).st_size
+ # Preserve executable IOU permissions even under a restrictive umask.
+ permissions = stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC
+ if hasattr(os, "fchmod"):
+ os.fchmod(descriptor, permissions)
+ else:
+ directory.verify()
+ os.chmod(os.path.join(directory.path, temporary), permissions)
+ os.close(descriptor)
+ descriptor = None
+ async with image_lock(image_path):
+ directory.verify()
+ if os.path.lexists(image_path):
+ raise InvalidImageError(
+ f"File '{image_path}' already exists, "
+ f"please choose a different name or remove the existing image"
+ )
+ checksum_str = checksum.hexdigest()
+ duplicate_image = await images_repo.get_image_by_checksum(checksum_str, os.path.dirname(image_path))
+ if duplicate_image:
+ raise InvalidImageError(
+ f"Image '{duplicate_image.filename}' with the same checksum "
+ f"already exists in '{os.path.dirname(image_path)}'"
+ )
+ file_fingerprint = directory.publish(temporary, image_name)
+ # Complete files survive a database failure for reconciliation.
+ image = await images_repo.save_verified_image(
+ dict(
+ image_name=image_name,
+ image_type=image_type,
+ image_size=image_size,
+ path=image_path,
+ checksum=checksum_str,
+ checksum_algorithm="md5",
+ file_fingerprint=file_fingerprint,
+ )
+ )
+ if image is None:
+ raise InvalidImageError(f"Failed to save image '{image_name}' to database")
+ return image
+ finally:
+ if descriptor is not None:
+ os.close(descriptor)
+ try:
+ directory.remove(temporary)
+ except FileNotFoundError:
+ pass
+ except OSError:
+ log.warning("Could not remove temporary image '%s'", temporary)
diff --git a/tests/api/routes/controller/test_image_subfolders.py b/tests/api/routes/controller/test_image_subfolders.py
new file mode 100644
index 000000000..ea82e8d79
--- /dev/null
+++ b/tests/api/routes/controller/test_image_subfolders.py
@@ -0,0 +1,254 @@
+"""Uploads into server-selected type folders and restricted relative subfolders."""
+
+import hashlib
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.exc import SQLAlchemyError
+
+from gns3server.controller.controller_error import ControllerError
+from gns3server.db.repositories.images import ImagesRepository
+from tests.api.routes.controller.test_image_sync import sync_service # noqa: F401 - pytest fixture discovery
+
+pytestmark = pytest.mark.asyncio
+QCOW = b"QFI\xfb\x00\x00\x00"
+QCOW_CHECKSUM = hashlib.md5(QCOW, usedforsecurity=False).hexdigest()
+
+
+async def upload(client, filename="router.qcow2", **kwargs):
+ return await client.post(f"/v3/images/upload/{filename}", content=kwargs.pop("content", QCOW), **kwargs)
+
+
+class TestImageSubfolders:
+ @pytest.mark.parametrize(
+ "content,folder,subfolder,filename",
+ [
+ (QCOW, "QEMU", "Vendor/Version 1", "router.bin"),
+ (b"\x7fELF\x01\x02\x01", "IOS", "Vendor/Version 1", "router.bin"),
+ (b"\x7fELF\x02\x01\x01", "IOU", "Vendor/Version 1", "router.bin"),
+ (QCOW, "QEMU", "TACACS", "tacacs.qcow2"),
+ ],
+ )
+ async def test_server_detects_type_and_sync_restores_nested_image(
+ self,
+ client,
+ images_dir,
+ db_session,
+ sync_service, # noqa: F811 - pytest fixture imported for discovery
+ content,
+ folder,
+ subfolder,
+ filename,
+ ):
+ response = await upload(
+ client, filename, params={"subdirectory": subfolder, "install_appliances": False}, content=content
+ )
+ assert response.status_code == 201, response.text
+ path = Path(images_dir) / folder / subfolder / filename
+ assert response.json()["path"] == str(path)
+ assert path.read_bytes() == content
+ assert not (Path(images_dir) / folder / filename).exists()
+ repository = ImagesRepository(db_session)
+ image_id = (await repository.get_image(str(path))).image_id
+ path.unlink()
+ await client.post("/v3/images/sync", json={})
+ await sync_service.task
+ assert (await repository.get_image(str(path), refresh=True)).availability == "missing"
+ path.write_bytes(content) # Same filesystem operation used by an SFTP copy.
+ sibling = path.with_name("copied-over-ssh.bin")
+ sibling.write_bytes(content + b"new")
+ await client.post("/v3/images/sync", json={})
+ await sync_service.task
+ restored = await repository.get_image(str(path), refresh=True)
+ assert restored.image_id == image_id
+ assert restored.availability == "available"
+ assert await repository.get_image(str(sibling)) is not None
+
+ @pytest.mark.parametrize(
+ "folder",
+ [
+ "../escape",
+ "/tmp/escape",
+ "C:/escape",
+ "C:\\escape",
+ "a/../b",
+ "a//b",
+ "a/",
+ ".hidden",
+ "a/%2e%2e",
+ "a/%252e%252e",
+ "a/\x00bad",
+ "a/CON",
+ "a/NUL.txt",
+ "lib",
+ "lib64",
+ "a/b.tmp",
+ "a/b.md5sum",
+ "a/b.",
+ "a/b ",
+ "/".join(["a"] * 9),
+ "a" * 65,
+ ],
+ )
+ async def test_rejects_unsafe_subfolders(self, client, images_dir, folder):
+ before = set(Path(images_dir).rglob("*"))
+ response = await upload(client, params={"subdirectory": folder})
+ assert response.status_code == 400, response.text
+ assert set(Path(images_dir).rglob("*")) == before
+
+ @pytest.mark.parametrize("filename", ["QEMU/router.qcow2", ".hidden", "NUL.qcow2", "router:stream.qcow2"])
+ async def test_subfolder_upload_requires_plain_filename(self, client, filename):
+ response = await upload(client, filename, params={"subdirectory": "Vendor"})
+ assert response.status_code == 400
+
+ @pytest.mark.parametrize("link,inside", [("QEMU/link", False), ("QEMU/link", True), ("QEMU", False)])
+ async def test_refuses_symlinks_without_creating_children(self, client, images_dir, tmp_path, link, inside):
+ target = Path(images_dir) / "safe-target" if inside else tmp_path / "outside"
+ target.mkdir()
+ link_path = Path(images_dir) / link
+ if link_path.is_dir():
+ link_path.rmdir()
+ link_path.symlink_to(target, target_is_directory=True)
+ folder = "link/child" if link.endswith("/link") else "Vendor/Version"
+ response = await upload(client, params={"subdirectory": folder})
+ assert response.status_code == 409
+ assert not list(target.iterdir())
+
+ async def test_duplicate_checks_are_scoped_to_destination(self, client, images_dir):
+ for folder in ("VendorA", "VendorB"):
+ response = await upload(
+ client, "same.qcow2", params={"subdirectory": folder}, headers={"X-MD5-Checksum": QCOW_CHECKSUM}
+ )
+ assert response.status_code == 201
+ response = await upload(client, "duplicate.qcow2", params={"subdirectory": "VendorA"})
+ assert response.status_code == 409
+ assert not (Path(images_dir) / "QEMU/VendorA/duplicate.qcow2").exists()
+ response = await upload(client, "same.qcow2", params={"subdirectory": "VendorA"}, content=QCOW + b"changed")
+ assert response.status_code == 409
+ assert (Path(images_dir) / "QEMU/VendorA/same.qcow2").read_bytes() == QCOW
+
+ async def test_empty_subfolder_keeps_default_location(self, client, images_dir):
+ response = await upload(client, "default.qcow2", params={"subdirectory": ""})
+ assert response.status_code == 201
+ assert response.json()["path"] == str(Path(images_dir) / "QEMU/default.qcow2")
+
+ async def test_subfolder_creation_requires_authenticated_allocate_permission(self, client, images_dir, test_user):
+ from gns3server.services import auth_service
+ from gns3server.services.authentication import DEFAULT_JWT_SECRET_KEY
+
+ response = await upload(client, params={"subdirectory": "Unauthorized"}, headers={"Authorization": ""})
+ assert response.status_code == 401
+ token = auth_service.create_access_token(test_user.username, secret_key=DEFAULT_JWT_SECRET_KEY)
+ with patch(
+ "gns3server.db.repositories.rbac.RbacRepository.check_user_has_privilege", new=AsyncMock(return_value=False)
+ ):
+ response = await upload(
+ client, params={"subdirectory": "Unauthorized"}, headers={"Authorization": f"Bearer {token}"}
+ )
+ assert response.status_code == 403
+ assert not (Path(images_dir) / "QEMU/Unauthorized").exists()
+
+
+@pytest.mark.parametrize(
+ "results", [[], [{"status": "created", "name": "Router"}], [{"status": "skipped", "reason": "Missing disk"}]]
+)
+async def test_upload_reports_template_outcomes(client, controller, images_dir, results):
+ with patch.object(
+ controller.appliance_manager, "install_appliances_from_image", new=AsyncMock(return_value=results)
+ ) as install:
+ response = await upload(client, "custom.qcow2", params={"install_appliances": True, "subdirectory": "TACACS"})
+ assert response.status_code == 201, response.text
+ assert Path(response.json()["path"]).read_bytes() == QCOW
+ install.assert_awaited_once()
+ assert install.call_args.args[1] == QCOW_CHECKSUM
+ if results:
+ assert response.json()["template_results"][0]["status"] == results[0]["status"]
+ else:
+ assert (
+ "No compatible appliance definition found in the server's catalog for this image."
+ in response.json()["template_results"][0]["reason"]
+ )
+
+
+@pytest.mark.parametrize(
+ "rollback,error",
+ [(False, ControllerError("Missing disk")), (True, SQLAlchemyError("Template transaction failed"))],
+ ids=["missing-dependency", "database-rollback"],
+)
+async def test_template_failure_preserves_uploaded_image(client, controller, images_dir, db_session, rollback, error):
+ async def fail(*args):
+ if rollback:
+ await db_session.rollback()
+ raise error
+
+ with patch.object(controller.appliance_manager, "install_appliances_from_image", side_effect=fail):
+ response = await upload(client, "custom.qcow2", params={"install_appliances": True})
+ assert response.status_code == 201, response.text
+ assert Path(response.json()["path"]).read_bytes() == QCOW
+ assert str(error) in response.json()["template_results"][0]["reason"]
+
+
+async def test_normal_upload_omits_template_results(client, images_dir):
+ response = await upload(client, "custom.qcow2")
+ assert response.status_code == 201, response.text
+ assert "template_results" not in response.json()
+
+
+async def test_pre_upload_compatibility_is_read_only(client, controller, images_dir):
+ with patch.object(
+ controller.appliance_manager,
+ "check_image_compatibility",
+ new=AsyncMock(return_value=[{"checksum": "a" * 32, "matches": []}]),
+ ) as check:
+ response = await client.post("/v3/images/compatibility", json={"checksums": ["a" * 32]})
+ assert response.status_code == 200, response.text
+ assert response.json() == [{"checksum": "a" * 32, "matches": []}]
+ check.assert_awaited_once()
+ assert list(Path(images_dir).rglob("*.qcow2")) == []
+
+
+@pytest.mark.parametrize(
+ "payload",
+ [
+ {"checksums": []},
+ {"checksums": ["../escape"]},
+ {"checksums": ["a" * 33]},
+ {"checksums": ["a" * 32] * 1001},
+ {"checksums": ["a" * 32], "path": "/tmp"},
+ ],
+)
+async def test_compatibility_rejects_invalid_input(client, payload):
+ response = await client.post("/v3/images/compatibility", json=payload)
+ assert response.status_code == 422
+
+
+async def test_catalog_size_filter_is_available_before_upload(client, controller):
+ with patch.object(
+ controller.appliance_manager,
+ "image_compatibility_catalog",
+ return_value={"image_sizes": [123], "has_unknown_sizes": False},
+ ):
+ response = await client.get("/v3/images/compatibility/catalog")
+ assert response.status_code == 200, response.text
+ assert response.json() == {"image_sizes": [123], "has_unknown_sizes": False}
+
+
+async def test_appliance_dependency_download_preserves_nested_destination(controller, images_dir, db_session):
+ from types import SimpleNamespace
+
+ async def content():
+ yield QCOW
+
+ response = SimpleNamespace(status=200, content=SimpleNamespace(iter_any=content))
+ context = AsyncMock()
+ context.__aenter__.return_value = response
+ destination = Path(images_dir) / "QEMU/Vendor/Version"
+ with patch("gns3server.controller.appliance_manager.HTTPClient.get", return_value=context):
+ image = await controller.appliance_manager._download_image(
+ str(destination), "dependency.qcow2", "qemu", "https://example.com/disk", ImagesRepository(db_session)
+ )
+ expected = destination / "dependency.qcow2"
+ assert image.path == str(expected)
+ assert expected.read_bytes() == QCOW
+ assert not (Path(images_dir) / "QEMU/dependency.qcow2").exists()
diff --git a/tests/api/routes/controller/test_templates.py b/tests/api/routes/controller/test_templates.py
index 932d0a9d4..d9c3f3e72 100644
--- a/tests/api/routes/controller/test_templates.py
+++ b/tests/api/routes/controller/test_templates.py
@@ -1582,3 +1582,67 @@ class TestImageAssociationWithTemplate:
response = await client.post(app.url_path_for("create_template"), json=params)
assert response.status_code == status.HTTP_404_NOT_FOUND
+
+ @pytest.mark.parametrize("reference", ["Vendor/router.bin", "router.bin"])
+ async def test_nested_template_resolves_image_in_correct_type_root(self, client, images_dir, db_session, reference):
+ repository = ImagesRepository(db_session)
+ # Insert the wrong type first, then a suffix collision below QEMU itself.
+ for index, relative in enumerate([f"IOU/{reference}", f"QEMU/Other/{reference}", f"QEMU/{reference}"]):
+ path = Path(images_dir) / relative
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(b"image bytes")
+ await repository.add_image(
+ path.name, "iou" if index == 0 else "qemu", 11, str(path), str(index) * 32, "md5"
+ )
+ expected_path = str(Path(images_dir) / "QEMU" / reference)
+ response = await client.post(
+ "/v3/templates",
+ json={
+ "name": f"Nested QEMU {reference}",
+ "template_type": "qemu",
+ "compute_id": "local",
+ "hda_disk_image": reference,
+ "ram": 512,
+ },
+ )
+ assert response.status_code == 201, response.text
+ templates = TemplatesRepository(db_session)
+ template = await templates.get_template(uuid.UUID(response.json()["template_id"]))
+ assert [image.path for image in template.images] == [expected_path]
+ # Updating removes the association for the same correctly resolved path.
+ response = await client.put(f"/v3/templates/{template.template_id}", json={"hda_disk_image": ""})
+ assert response.status_code == 200, response.text
+ await db_session.refresh(template, ["images"])
+ assert template.images == []
+
+ async def test_update_removes_legacy_association_when_new_exact_path_exists(self, client, images_dir, db_session):
+ repository = ImagesRepository(db_session)
+ legacy_path = Path(images_dir) / "QEMU/Other/Vendor/router.qcow2"
+ canonical_path = Path(images_dir) / "QEMU/Vendor/router.qcow2"
+ for index, path in enumerate([legacy_path, canonical_path]):
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(b"image bytes")
+ await repository.add_image(path.name, "qemu", 11, str(path), str(index) * 32, "md5")
+ response = await client.post(
+ "/v3/templates",
+ json={
+ "name": "Legacy nested association",
+ "template_type": "qemu",
+ "compute_id": "local",
+ "hda_disk_image": str(legacy_path),
+ "ram": 512,
+ },
+ )
+ assert response.status_code == 201, response.text
+ templates = TemplatesRepository(db_session)
+ template = await templates.get_template(uuid.UUID(response.json()["template_id"]))
+ # Simulate an existing template that used the legacy suffix lookup.
+ await templates.update_template(template, {"hda_disk_image": "Vendor/router.qcow2"})
+ legacy_image = template.images[0]
+ legacy_image.availability = "missing"
+ await db_session.commit()
+ legacy_path.unlink()
+ response = await client.put(f"/v3/templates/{template.template_id}", json={"hda_disk_image": ""})
+ assert response.status_code == 200, response.text
+ await db_session.refresh(template, ["images"])
+ assert template.images == []
diff --git a/tests/controller/test_appliance_manager.py b/tests/controller/test_appliance_manager.py
index 0700b0312..8725c6d04 100644
--- a/tests/controller/test_appliance_manager.py
+++ b/tests/controller/test_appliance_manager.py
@@ -15,7 +15,12 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
+import copy
+import json
import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
import pytest
@@ -160,3 +165,67 @@ async def test_install_version_not_found(monkeypatch):
with pytest.raises(ControllerNotFoundError):
await manager.install_appliance(uuid.UUID(appliance.id), "9.9", None, None, None, None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "dependency_state",
+ ["missing", "selected", "existing", "downloadable"],
+)
+async def test_image_compatibility_accounts_for_dependencies_without_installing(dependency_state):
+ data = json.loads(Path("gns3server/appliances/empty-vm.gns3a").read_text())
+ primary, dependency = data["images"][:2]
+ if dependency_state != "downloadable":
+ dependency.pop("direct_download_url")
+ data["images"] = [primary, dependency]
+ data["versions"] = [
+ {
+ "name": primary["version"],
+ "images": {"hda_disk_image": primary["filename"], "hdb_disk_image": dependency["filename"]},
+ }
+ ]
+ original = copy.deepcopy(data)
+ appliance = Appliance("test.gns3a", data)
+ manager = ApplianceManager()
+ manager._appliances[appliance.id] = appliance
+ repo = AsyncMock()
+ repo.get_image_by_checksum.return_value = object() if dependency_state == "existing" else None
+ manager._download_image = AsyncMock()
+ manager._create_template = AsyncMock()
+ checksums = [primary["md5sum"]]
+ if dependency_state == "selected":
+ checksums.append(dependency["md5sum"])
+ result = await manager.check_image_compatibility(checksums, repo)
+ match = result[0]["matches"][0]
+ assert match["name"] == "Empty VM"
+ assert match["version"] == primary["version"]
+ assert match["missing_images"] == ([dependency["filename"]] if dependency_state == "missing" else [])
+ assert match["downloadable_images"] == ([dependency["filename"]] if dependency_state == "downloadable" else [])
+ manager._download_image.assert_not_awaited()
+ manager._create_template.assert_not_awaited()
+ assert appliance._data == original
+
+
+@pytest.mark.asyncio
+async def test_image_compatibility_returns_no_matches_for_custom_checksum():
+ manager = ApplianceManager()
+ repo = AsyncMock()
+ result = await manager.check_image_compatibility(["A" * 32, "a" * 32], repo)
+ assert result == [{"checksum": "a" * 32, "matches": []}]
+ repo.get_image_by_checksum.assert_not_awaited()
+
+
+def test_image_catalog_filter_is_conservative_and_does_not_match_by_filename():
+ manager = ApplianceManager()
+ manager._appliances = {
+ "a": SimpleNamespace(
+ images=[
+ {"md5sum": "a" * 32, "filesize": 123, "filename": "router.qcow2"},
+ {"md5sum": "b" * 32, "filesize": 123},
+ {"filesize": 456},
+ ]
+ )
+ }
+ assert manager.image_compatibility_catalog() == {"image_sizes": [123], "has_unknown_sizes": False}
+ manager._appliances["a"].images.append({"md5sum": "c" * 32})
+ assert manager.image_compatibility_catalog()["has_unknown_sizes"] is True
diff --git a/tests/services/test_image_reconciliation.py b/tests/services/test_image_reconciliation.py
index 8ee12856d..ff90dd186 100644
--- a/tests/services/test_image_reconciliation.py
+++ b/tests/services/test_image_reconciliation.py
@@ -707,3 +707,131 @@ async def test_delete_preserves_replacement_created_while_waiting_for_lock(inven
assert await repository.prune_images() == 0
assert path.read_bytes() == replacement
assert (await rows(inventory))[0]["image_id"] == replacement_id
+
+
+async def upload_stream(inventory, config, stream, *, subdirectory, filename="router.qcow2"):
+ async with AsyncSession(inventory.engine) as session:
+ return await write_image(
+ filename,
+ str(Path(config.settings.Server.images_path) / filename),
+ stream,
+ ImagesRepository(session),
+ subdirectory=subdirectory,
+ )
+
+
+async def test_upload_rejects_directory_swap_and_cleans_anchored_temporary(inventory, config, tmp_path):
+ if os.open not in os.supports_dir_fd:
+ pytest.skip("Requires directory descriptor support")
+ root = Path(config.settings.Server.images_path)
+ folder = root / "QEMU" / "vendor"
+ moved = root / "QEMU" / "original-vendor"
+ outside = tmp_path / "outside"
+ outside.mkdir()
+
+ async def swapping_stream():
+ yield QCOW
+ folder.rename(moved)
+ folder.symlink_to(outside, target_is_directory=True)
+ yield b"more bytes"
+
+ with pytest.raises(OSError):
+ await upload_stream(inventory, config, swapping_stream(), subdirectory="vendor")
+ assert not list(outside.iterdir())
+ assert not list(moved.iterdir())
+ assert not await rows(inventory)
+
+
+async def test_cancelled_nested_upload_cleans_temporary(inventory, config):
+ root = Path(config.settings.Server.images_path)
+
+ async def cancelled_stream():
+ yield QCOW
+ raise asyncio.CancelledError
+
+ with pytest.raises(asyncio.CancelledError):
+ await upload_stream(inventory, config, cancelled_stream(), subdirectory="Vendor/Version")
+ assert not list((root / "QEMU/Vendor/Version").iterdir())
+ assert not await rows(inventory)
+
+
+async def test_nested_appliance_reference_selects_correct_same_named_image(inventory, config):
+ from gns3server.controller.appliance_manager import ApplianceManager
+ from gns3server.db.repositories.templates import TemplatesRepository
+
+ image_file(config, "QEMU/other/router.qcow2", QCOW + b"other")
+ path = image_file(config, "QEMU/vendor/router.qcow2", QCOW + b"required")
+ await scan(inventory)
+ checksum = hashlib.md5(path.read_bytes(), usedforsecurity=False).hexdigest()
+ appliance = SimpleNamespace(images=[{"filename": path.name, "md5sum": checksum}])
+ version = {"images": {"hda_disk_image": path.name}}
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ await ApplianceManager()._find_appliance_version_images(
+ appliance, version, ImagesRepository(session), str(path.parent)
+ )
+ reference = version["images"]["hda_disk_image"]
+ assert reference == "vendor/router.qcow2"
+ resolved = await TemplatesRepository(session).get_image(reference)
+ assert resolved.path == str(path)
+ assert resolved.checksum == checksum
+
+
+@pytest.mark.parametrize("anchored", [True, False])
+async def test_upload_directory_publication_is_no_overwrite_in_both_modes(config, anchored):
+ from gns3server.utils.image_inventory import ImageUploadDirectory
+
+ root = Path(config.settings.Server.images_path)
+ directory = ImageUploadDirectory(str(root), str(root / "QEMU/Vendor/Version"))
+ if anchored and not directory.anchored:
+ pytest.skip("Requires directory descriptor support")
+ directory.anchored = anchored
+ with directory:
+ fd, temporary = directory.temporary()
+ os.write(fd, QCOW)
+ os.close(fd)
+ directory.publish(temporary, "router.qcow2")
+ fd, temporary = directory.temporary()
+ os.write(fd, QCOW + b"replacement")
+ os.close(fd)
+ with pytest.raises(FileExistsError):
+ directory.publish(temporary, "router.qcow2")
+ directory.remove(temporary)
+ assert (root / "QEMU/Vendor/Version/router.qcow2").read_bytes() == QCOW
+
+
+@pytest.mark.parametrize("anchored", [True, False])
+@pytest.mark.parametrize("inside", [True, False])
+async def test_upload_directory_rejects_symlink_before_creating_children(config, tmp_path, anchored, inside):
+ from gns3server.utils.image_inventory import ImageUploadDirectory
+
+ root = Path(config.settings.Server.images_path)
+ (root / "QEMU").mkdir(parents=True, exist_ok=True)
+ target = root / "target" if inside else tmp_path / "outside"
+ target.mkdir()
+ (root / "QEMU/link").symlink_to(target, target_is_directory=True)
+ directory = ImageUploadDirectory(str(root), str(root / "QEMU/link/child"))
+ if anchored and not directory.anchored:
+ pytest.skip("Requires directory descriptor support")
+ directory.anchored = anchored
+ with pytest.raises(OSError), directory:
+ pytest.fail("A symlink folder must not be accepted")
+ assert not list(target.iterdir())
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits and umask")
+async def test_upload_restores_executable_permissions_under_restrictive_umask(inventory, config):
+ import stat
+
+ root = Path(config.settings.Server.images_path)
+ (root / "IOU/Vendor").mkdir(parents=True)
+ (Path(config.config_dir) / ".image-locks").mkdir(parents=True, exist_ok=True)
+
+ async def stream():
+ yield b"\x7fELF\x02\x01\x01"
+
+ previous_umask = os.umask(0o777)
+ try:
+ image = await upload_stream(inventory, config, stream(), filename="router.bin", subdirectory="Vendor")
+ finally:
+ os.umask(previous_umask)
+ assert stat.S_IMODE(os.stat(image.path).st_mode) == 0o700