diff --git a/docs/README.md b/docs/README.md
index ea1581292..6c0887b2a 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -59,6 +59,16 @@ docs/
## Features
+### Image inventory synchronization
+Image Manager's **Sync Images** action reconciles the catalog with the configured
+image directories. Automatic synchronization uses filesystem events and periodic
+scans (`auto_discover_images`, `image_sync_interval`, default 900 seconds).
+New and changed images are checksummed; missing records retain their IDs and
+template associations and become available again when restored at the same path.
+The `/v3/images/sync` API supports dry-run and forced checksum verification.
+Database migration `d9e8a2b7c401` is required when upgrading an existing catalog.
+Atomic image publication requires filesystem hard-link support.
+
### Controller + Compute Setup (`features/compute-controller-setup.md`)
Architecture and minimum configuration for setting up GNS3 Controller with remote Compute nodes. Covers compute node config, controller registration, and multi-compute deployment.
diff --git a/gns3server/api/routes/controller/images.py b/gns3server/api/routes/controller/images.py
index 65f534b8e..0496e29ee 100644
--- a/gns3server/api/routes/controller/images.py
+++ b/gns3server/api/routes/controller/images.py
@@ -21,12 +21,14 @@ API routes for images.
import os
import logging
import urllib.parse
+import tempfile
-from fastapi import APIRouter, Request, Depends, status
+from fastapi import APIRouter, Request, Response, Depends, Query, status
from fastapi.encoders import jsonable_encoder
from starlette.requests import ClientDisconnect
from sqlalchemy.orm.exc import MultipleResultsFound
-from typing import List, Optional
+from sqlalchemy.exc import SQLAlchemyError
+from typing import List, Optional, Literal
from gns3server import schemas
from gns3server.config import Config
@@ -43,6 +45,8 @@ from gns3server.db.repositories.images import ImagesRepository
from gns3server.db.repositories.templates import TemplatesRepository
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.controller import Controller
+from gns3server.services.image_reconciliation import get_image_reconciliation_service
+from gns3server.utils.image_inventory import contained_path, image_lock, publish_image, fingerprint, ImageLockBusy
from gns3server.controller.controller_error import (
ControllerError,
ControllerNotFoundError,
@@ -59,6 +63,43 @@ log = logging.getLogger(__name__)
router = APIRouter()
+def image_destination(image_path):
+ root = os.path.realpath(os.path.expanduser(Config.instance().settings.Server.images_path))
+ full_path = os.path.abspath(os.path.join(root, image_path))
+ if not contained_path(os.path.realpath(full_path), root):
+ raise ControllerForbiddenError(f"Cannot write image, '{image_path}' is forbidden")
+ return full_path
+
+
+@router.post(
+ "/sync",
+ response_model=schemas.ImageSyncJob,
+ status_code=status.HTTP_202_ACCEPTED,
+ dependencies=[Depends(has_privilege("Image.Allocate"))],
+)
+async def sync_images(options: schemas.ImageSyncRequest, request: Request, response: Response):
+ """Reconcile configured image directories without deleting files or references."""
+ try:
+ job = await get_image_reconciliation_service(request.app).start(**options.model_dump())
+ except ImageLockBusy:
+ raise ControllerError("Image synchronization is already running or shutting down")
+ response.headers["Location"] = str(request.url_for("get_image_sync_job", job_id=job["job_id"]))
+ return job
+
+
+@router.get(
+ "/sync/jobs/{job_id}", response_model=schemas.ImageSyncJob, dependencies=[Depends(has_privilege("Image.Audit"))]
+)
+async def get_image_sync_job(
+ job_id: str, request: Request, offset: int = Query(0, ge=0), limit: int = Query(100, ge=1, le=1000)
+):
+ """Get progress and a page of errors (first 1,000 errors retained per job)."""
+ job = await get_image_reconciliation_service(request.app).get_job(job_id, offset, limit)
+ if job is None:
+ raise ControllerNotFoundError(f"Image synchronization job '{job_id}' not found")
+ return job
+
+
@router.post(
"/qemu/{image_path:path}",
response_model=schemas.Image,
@@ -69,7 +110,7 @@ async def create_qemu_image(
image_path: str,
image_data: schemas.QemuDiskImageCreate,
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
-) -> models.Image:
+) -> Optional[models.Image]:
"""
Create a new blank Qemu image.
@@ -83,39 +124,41 @@ async def create_qemu_image(
disk_image_path = urllib.parse.unquote(image_path)
image_dir, image_name = os.path.split(disk_image_path)
# check if the path is within the default images directory
- base_images_directory = os.path.expanduser(Config.instance().settings.Server.images_path)
- full_path = os.path.abspath(os.path.join(base_images_directory, image_dir, image_name))
- if os.path.commonprefix([base_images_directory, full_path]) != base_images_directory:
- raise ControllerForbiddenError(f"Cannot write disk image, '{disk_image_path}' is forbidden")
+ disk_image_path = image_destination(disk_image_path)
if not image_dir:
# put the image in the default images directory for Qemu
directory = default_images_directory(image_type="qemu")
os.makedirs(directory, exist_ok=True)
- disk_image_path = os.path.abspath(os.path.join(directory, disk_image_path))
+ disk_image_path = image_destination(os.path.join(directory, image_name))
- if await images_repo.get_image(disk_image_path):
- raise ControllerBadRequestError(f"Disk image '{disk_image_path}' already exists")
-
- options = jsonable_encoder(image_data, exclude_unset=True)
- # FIXME: should we have the create_disk_image in the compute code since
- # this code is used to create images on the controller?
- await Qemu.instance().create_disk_image(disk_image_path, options)
-
- image_info = await read_image_info(disk_image_path, "qemu")
-
- image = await images_repo.get_image(disk_image_path)
- if image:
- # the image has already been added to the database
- return image
- else:
- return await images_repo.add_image(**image_info)
+ async with image_lock(disk_image_path):
+ if os.path.lexists(disk_image_path):
+ raise ControllerBadRequestError(f"Disk image '{disk_image_path}' already exists")
+ os.makedirs(os.path.dirname(disk_image_path), exist_ok=True)
+ fd, temporary = tempfile.mkstemp(prefix=".gns3-create-", suffix=".tmp", dir=os.path.dirname(disk_image_path))
+ os.close(fd)
+ try:
+ options = jsonable_encoder(image_data, exclude_unset=True)
+ await Qemu.instance().create_disk_image(temporary, options)
+ image_info = await read_image_info(temporary, "qemu", allow_raw_image=allow_raw_image)
+ publish_image(temporary, disk_image_path)
+ image_info.update(
+ path=disk_image_path, image_name=image_name, file_fingerprint=fingerprint(disk_image_path)
+ )
+ return await images_repo.save_verified_image(image_info)
+ except (OSError, InvalidImageError, SQLAlchemyError) as e:
+ raise ControllerError(f"Could not create disk image '{disk_image_path}': {e}") from e
+ finally:
+ if os.path.exists(temporary):
+ os.unlink(temporary)
@router.get("", response_model=List[schemas.Image], dependencies=[Depends(has_privilege("Image.Audit"))])
async def get_images(
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
image_type: Optional[schemas.ImageType] = None,
+ availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None,
) -> List[models.Image]:
"""
Return all images.
@@ -123,7 +166,7 @@ async def get_images(
Required privilege: Image.Audit
"""
- return await images_repo.get_images(image_type)
+ return await images_repo.get_images(image_type, availability)
@router.post(
@@ -153,10 +196,7 @@ async def upload_image(
image_path = urllib.parse.unquote(image_path)
image_dir, image_name = os.path.split(image_path)
# check if the path is within the default images directory
- base_images_directory = os.path.expanduser(Config.instance().settings.Server.images_path)
- full_path = os.path.abspath(os.path.join(base_images_directory, image_dir, image_name))
- if os.path.commonprefix([base_images_directory, full_path]) != base_images_directory:
- raise ControllerForbiddenError(f"Cannot write image, '{image_path}' is forbidden")
+ 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")
@@ -170,7 +210,10 @@ 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)
- except (OSError, InvalidImageError, ClientDisconnect) as e:
+ except (OSError, InvalidImageError, ClientDisconnect, SQLAlchemyError) as e:
+ service = getattr(request.app.state, "image_reconciliation", None)
+ if service:
+ service.dirty.set()
raise ControllerError(f"Could not save image '{image_path}': {e}")
if install_appliances:
@@ -206,7 +249,9 @@ async def prune_images(
# a single pass over all projects' node properties protects every
# referenced file name at once
referenced_filenames = Controller.instance().collect_referenced_image_filenames()
- await images_repo.prune_images(list(skip_images) + list(referenced_filenames))
+ await images_repo.prune_images(
+ list(skip_images) + list(referenced_filenames), is_in_use=Controller.instance().find_projects_using_image
+ )
@router.post("/install", status_code=status.HTTP_200_OK, dependencies=[Depends(has_privilege("Image.Allocate"))])
@@ -228,6 +273,11 @@ async def install_images(
skip_images = get_builtin_disks()
images = await images_repo.get_images()
for image in images:
+ if not await images_repo.is_usable(image):
+ skipped.append(
+ {"name": image.filename, "reason": "image is missing, changed or unreadable; synchronize images first"}
+ )
+ continue
if skip_images and image.filename in skip_images:
log.debug(f"Skipping image '{image.path}' for image installation")
continue
@@ -307,11 +357,22 @@ async def delete_image(
if project_names:
raise ControllerError(f"Image '{image_path}' is used by one or more projects: {', '.join(project_names)}")
- try:
- os.remove(image.path)
- except OSError:
- log.warning(f"Could not delete image file {image.path}")
-
- success = await images_repo.delete_image(image_path)
- if not success:
- raise ControllerError(f"Image '{image_path}' could not be deleted")
+ path = image.path
+ revision = (image.image_id, image.checksum, image.file_fingerprint)
+ async with image_lock(path):
+ image = await images_repo.get_image(path, refresh=True)
+ if image is None or (image.image_id, image.checksum, image.file_fingerprint) != revision:
+ raise ControllerError(f"Image '{image_path}' changed while waiting for deletion; refresh and retry")
+ # Recheck usage after waiting for a concurrent writer/scanner.
+ if await images_repo.get_image_templates(image.image_id) or Controller.instance().find_projects_using_image(
+ image.filename
+ ):
+ raise ControllerError(f"Image '{image_path}' is in use")
+ try:
+ os.remove(image.path)
+ except FileNotFoundError:
+ pass
+ except OSError as e:
+ raise ControllerError(f"Could not delete image file '{image.path}': {e}") from e
+ if not await images_repo.delete_image_exact(image.image_id):
+ raise ControllerError(f"Image '{image_path}' could not be deleted")
diff --git a/gns3server/controller/appliance_manager.py b/gns3server/controller/appliance_manager.py
index 53dacadcf..e99320202 100644
--- a/gns3server/controller/appliance_manager.py
+++ b/gns3server/controller/appliance_manager.py
@@ -35,6 +35,7 @@ from ..utils.http_client import HTTPClient
from .controller_error import ControllerBadRequestError, ControllerNotFoundError, ControllerError
from .appliance_to_template import ApplianceToTemplate
from ..utils.images import InvalidImageError, write_image, read_image_info
+from ..utils.image_inventory import image_lock
from gns3server import schemas
from gns3server.schemas.controller.appliances import ApplianceModel
@@ -178,13 +179,16 @@ class ApplianceManager:
# check if the image is on disk but it not yet in the database
image_path = os.path.join(image_dir, appliance_file)
if os.path.exists(image_path):
- image_info = await read_image_info(image_path)
- if image_info.get("checksum") == image_checksum:
- log.info(f"Adding image '{image_path}' to the database")
+ async with image_lock(image_path):
+ image_info = await read_image_info(image_path, allow_raw_image=True)
+ if image_info["checksum"] != image_checksum:
+ raise ControllerError(
+ f"Image '{image_path}' does not match the appliance checksum"
+ )
try:
- await images_repo.add_image(**image_info)
+ await images_repo.save_verified_image(image_info)
except SQLAlchemyError as e:
- log.warning(f"Error while adding image '{image['path']}' to the database: {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")
diff --git a/gns3server/core/tasks.py b/gns3server/core/tasks.py
index 427fc2dd8..71d15bbb4 100644
--- a/gns3server/core/tasks.py
+++ b/gns3server/core/tasks.py
@@ -26,22 +26,23 @@ from gns3server.compute import MODULES
from gns3server.compute.port_manager import PortManager
from gns3server.compute.marker.marker_manager import MarkerManager
from gns3server.utils.http_client import HTTPClient
-from gns3server.db.tasks import connect_to_db, get_computes, disconnect_from_db, discover_images_on_filesystem
+from gns3server.db.tasks import connect_to_db, get_computes, disconnect_from_db
+from gns3server.services.image_reconciliation import get_image_reconciliation_service
import logging
log = logging.getLogger(__name__)
-auto_discover_images_task_handle = None
-
@asynccontextmanager
async def lifespan(app: FastAPI):
await startup(app)
- yield
- await shutdown(app)
+ try:
+ yield
+ finally:
+ await shutdown(app)
async def startup(app: FastAPI) -> None:
@@ -66,18 +67,7 @@ async def startup(app: FastAPI) -> None:
await Controller.instance().start(computes)
- # Because with a large image collection
- # without md5sum already computed we start the
- # computing with server start
- from gns3server.compute.qemu import Qemu
-
- if Config.instance().settings.Server.auto_discover_images is True:
- # Start the discovering new images on file system 5 seconds after the server has started
- # to give it a chance to process API requests
- global auto_discover_images_task_handle
- auto_discover_images_task_handle = asyncio.get_event_loop().call_later(
- 5, lambda: asyncio.create_task(discover_images_on_filesystem(app))
- )
+ get_image_reconciliation_service(app).start_background()
for module in MODULES:
log.debug(f"Loading module {module.__name__}")
@@ -107,8 +97,10 @@ async def shutdown(app: FastAPI) -> None:
Tasks to be performed when the server is exiting.
"""
- if auto_discover_images_task_handle is not None and not auto_discover_images_task_handle.cancelled():
- auto_discover_images_task_handle.cancel()
+ service = getattr(app.state, "image_reconciliation", None)
+ if service is not None:
+ await service.close()
+ del app.state.image_reconciliation
await HTTPClient.close_session()
await MarkerManager.instance().stop()
# Kill resident sharkd sessions (marker replay) and drop their /tmp
diff --git a/gns3server/db/models/__init__.py b/gns3server/db/models/__init__.py
index cd538b80d..e5e234be4 100644
--- a/gns3server/db/models/__init__.py
+++ b/gns3server/db/models/__init__.py
@@ -21,7 +21,7 @@ from .users import User, UserGroup
from .roles import Role
from .privileges import Privilege
from .computes import Compute
-from .images import Image
+from .images import Image, ImageSyncJob
from .pools import Resource, ResourcePool
from .llm_model_configs import LLMModelConfig
from .api_keys import ApiKey
diff --git a/gns3server/db/models/images.py b/gns3server/db/models/images.py
index 626773aad..d5b323564 100644
--- a/gns3server/db/models/images.py
+++ b/gns3server/db/models/images.py
@@ -15,7 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from sqlalchemy import Table, Column, String, ForeignKey, BigInteger, Integer
+from sqlalchemy import Table, Column, String, ForeignKey, BigInteger, Integer, DateTime, JSON, Boolean
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base import Base, BaseTable, GUID
@@ -39,4 +39,21 @@ class Image(BaseTable):
image_size = Column(BigInteger)
checksum = Column(String, index=True)
checksum_algorithm = Column(String)
+ availability = Column(String, nullable=False, default="unknown", server_default="unknown")
+ file_fingerprint = Column(String)
+ last_seen_at = Column(DateTime)
+ last_verified_at = Column(DateTime)
+ last_error = Column(String)
templates = relationship("Template", secondary=image_template_map, back_populates="images")
+
+
+class ImageSyncJob(BaseTable):
+ __tablename__ = "image_sync_jobs"
+
+ job_id = Column(String, primary_key=True)
+ status = Column(String, nullable=False)
+ dry_run = Column(Boolean, nullable=False)
+ force_checksum = Column(Boolean, nullable=False)
+ finished_at = Column(DateTime)
+ counts = Column(JSON, nullable=False)
+ errors = Column(JSON, nullable=False)
diff --git a/gns3server/db/repositories/images.py b/gns3server/db/repositories/images.py
index 5f9b35520..1d52295e7 100644
--- a/gns3server/db/repositories/images.py
+++ b/gns3server/db/repositories/images.py
@@ -16,11 +16,15 @@
# along with this program. If not, see .
import os
+import asyncio
+from datetime import datetime, timezone
-from typing import Optional, List, cast
+from typing import Optional, List, Callable, cast
from sqlalchemy import select, delete, update
from sqlalchemy.engine import CursorResult
from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
+from gns3server.utils.image_inventory import fingerprint, image_lock, normalized_path
from .base import BaseRepository
@@ -36,19 +40,21 @@ class ImagesRepository(BaseRepository):
super().__init__(db_session)
- async def get_image(self, image_path: str) -> Optional[models.Image]:
+ async def get_image(self, image_path: str, *, refresh: bool = False) -> Optional[models.Image]:
"""
Get an image by its path.
"""
image_dir, image_name = os.path.split(image_path)
- if image_dir:
+ 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(image_path)
+ models.Image.filename == image_name, models.Image.path.endswith(os.sep + image_path, autoescape=True)
)
else:
query = select(models.Image).where(models.Image.filename == image_name)
- result = await self._db_session.execute(query)
+ result = await self._db_session.execute(query.execution_options(populate_existing=refresh))
return result.scalars().one_or_none()
async def get_image_by_checksum(self, checksum: str, image_dir: Optional[str] = None) -> Optional[models.Image]:
@@ -56,18 +62,29 @@ class ImagesRepository(BaseRepository):
Get an image by its checksum.
"""
- if image_dir:
- query = select(models.Image).where(
- models.Image.checksum == checksum, models.Image.path.startswith(image_dir)
- )
- result = await self._db_session.execute(query)
- return result.scalars().one_or_none()
- else:
- query = select(models.Image).where(models.Image.checksum == checksum)
- result = await self._db_session.execute(query)
- return result.scalars().first()
+ query = select(models.Image).where(models.Image.checksum == checksum).order_by(models.Image.image_id)
+ result = await self._db_session.execute(query)
+ for image in result.scalars().all():
+ if image_dir and normalized_path(os.path.dirname(image.path)) != normalized_path(image_dir):
+ continue
+ if await self.is_usable(image):
+ return image
+ return None
- async def get_images(self, image_type=None) -> List[models.Image]:
+ async def is_usable(self, image: models.Image) -> bool:
+ """Validate a checksum candidate without trusting stale catalog/sidecar data."""
+ from gns3server.utils.images import inspect_image_file, InvalidImageError
+
+ try:
+ current = await asyncio.to_thread(fingerprint, image.path)
+ if image.availability == "available" and image.file_fingerprint == current:
+ return True
+ info = await asyncio.to_thread(inspect_image_file, image.path, image.image_type, True)
+ return info["checksum"] == image.checksum and info["image_size"] == image.image_size
+ except (OSError, InvalidImageError):
+ return False
+
+ async def get_images(self, image_type=None, availability=None) -> List[models.Image]:
"""
Get all images.
"""
@@ -76,6 +93,8 @@ class ImagesRepository(BaseRepository):
query = select(models.Image).where(models.Image.image_type == image_type)
else:
query = select(models.Image)
+ if availability:
+ query = query.where(models.Image.availability == availability)
result = await self._db_session.execute(query)
return list(result.scalars().all())
@@ -89,7 +108,9 @@ class ImagesRepository(BaseRepository):
result = await self._db_session.execute(query)
return list(result.scalars().all())
- async def add_image(self, image_name, image_type, image_size, path, checksum, checksum_algorithm) -> models.Image:
+ async def add_image(
+ self, image_name, image_type, image_size, path, checksum, checksum_algorithm, file_fingerprint=None
+ ) -> models.Image:
"""
Create a new image.
"""
@@ -102,10 +123,16 @@ class ImagesRepository(BaseRepository):
path=path,
checksum=checksum,
checksum_algorithm=checksum_algorithm,
+ file_fingerprint=file_fingerprint,
+ availability="available" if file_fingerprint else "unknown",
)
self._db_session.add(db_image)
- await self._db_session.commit()
+ try:
+ await self._db_session.commit()
+ except Exception:
+ await self._db_session.rollback()
+ raise
await self._db_session.refresh(db_image)
return db_image
@@ -127,16 +154,91 @@ class ImagesRepository(BaseRepository):
await self._db_session.refresh(image_db) # force refresh of updated_at value
return image_db
+ async def save_verified_image(self, info: dict) -> Optional[models.Image]:
+ """Upsert an exact path, preserving template associations and the image ID.
+
+ Callers coordinate publication/inspection with image_lock(). Each commit
+ is short, and a concurrent insert from a legacy caller is retried safely.
+ """
+ values = dict(info)
+ values["filename"] = values.pop("image_name")
+ values.update(
+ availability="available",
+ last_error=None,
+ last_seen_at=datetime.now(timezone.utc).replace(tzinfo=None),
+ last_verified_at=datetime.now(timezone.utc).replace(tzinfo=None),
+ )
+ for attempt in range(2):
+ try:
+ image = (
+ await self._db_session.execute(
+ select(models.Image)
+ .where(models.Image.path == info["path"])
+ .execution_options(populate_existing=True)
+ )
+ ).scalar_one_or_none()
+ if image is None:
+ # A legacy spelling (e.g. /images/QEMU/./disk.qcow2) can
+ # refer to the same destination restored by an API upload.
+ candidates = (
+ (
+ await self._db_session.execute(
+ select(models.Image).where(models.Image.filename == values["filename"])
+ )
+ )
+ .scalars()
+ .all()
+ )
+ aliases = [
+ candidate
+ for candidate in candidates
+ if normalized_path(candidate.path) == normalized_path(info["path"])
+ ]
+ if len(aliases) > 1:
+ from sqlalchemy.exc import MultipleResultsFound
+
+ raise MultipleResultsFound("Ambiguous image path aliases; manual review required")
+ if aliases:
+ image = aliases[0]
+ values["path"] = image.path
+ if image is None:
+ image = models.Image(**values)
+ self._db_session.add(image)
+ else:
+ for key, value in values.items():
+ setattr(image, key, value)
+ await self._db_session.commit()
+ await self._db_session.refresh(image)
+ return image
+ except IntegrityError:
+ await self._db_session.rollback()
+ if attempt:
+ raise
+ except Exception:
+ await self._db_session.rollback()
+ raise
+ return None
+
+ async def delete_image_exact(self, image_id: int) -> bool:
+ result = await self._db_session.execute(delete(models.Image).where(models.Image.image_id == image_id))
+ await self._db_session.commit()
+ return cast(CursorResult, result).rowcount > 0
+
async def delete_image(self, image_path: str) -> bool:
"""
Delete an image.
"""
image_dir, image_name = os.path.split(image_path)
- if image_dir:
+ if os.path.isabs(image_path):
+ query = delete(models.Image).where(models.Image.path == image_path)
+ elif image_dir:
query = (
delete(models.Image)
- .where(models.Image.filename == image_name, models.Image.path.endswith(image_path))
+ .where(
+ models.Image.filename == image_name,
+ models.Image.path.endswith(os.sep + image_path, autoescape=True),
+ )
.execution_options(synchronize_session=False)
)
else:
@@ -145,25 +247,45 @@ class ImagesRepository(BaseRepository):
await self._db_session.commit()
return cast(CursorResult, result).rowcount > 0
- async def prune_images(self, skip_images: Optional[list[str]] = None) -> int:
+ async def prune_images(self, skip_images: Optional[list[str]] = None, is_in_use: Optional[Callable] = None) -> int:
"""
Prune images not attached to any template.
"""
query = select(models.Image).filter(~models.Image.templates.any())
result = await self._db_session.execute(query)
- images = result.scalars().all()
+ # Snapshot scalar values; commits can expire ORM instances.
+ images = [
+ (image.image_id, image.filename, image.path, image.checksum, image.file_fingerprint)
+ for image in result.scalars().all()
+ ]
images_deleted = 0
- for image in images:
- if skip_images and image.filename in skip_images:
- log.debug(f"Skipping image '{image.path}' for pruning")
+ errors = []
+ for image_id, filename, path, checksum, file_fingerprint in images:
+ if skip_images and filename in skip_images:
continue
- try:
- log.debug(f"Deleting image '{image.path}'")
- os.remove(image.path)
- except OSError:
- log.warning(f"Could not delete image file {image.path}")
- if await self.delete_image(image.path):
- images_deleted += 1
+ async with image_lock(path):
+ current = await self.get_image(path, refresh=True)
+ if current is None or (current.image_id, current.checksum, current.file_fingerprint) != (
+ image_id,
+ checksum,
+ file_fingerprint,
+ ):
+ continue # A concurrent request removed or replaced this image.
+ if await self.get_image_templates(image_id) or (is_in_use and is_in_use(filename)):
+ continue
+ try:
+ os.remove(path)
+ except FileNotFoundError:
+ pass
+ except OSError as e:
+ errors.append(f"{path}: {e}")
+ continue
+ if await self.delete_image_exact(image_id):
+ images_deleted += 1
log.info(f"{images_deleted} image(s) have been deleted")
+ if errors:
+ from gns3server.controller.controller_error import ControllerError
+
+ raise ControllerError("Could not delete image files: " + "; ".join(errors))
return images_deleted
diff --git a/gns3server/db/repositories/templates.py b/gns3server/db/repositories/templates.py
index 9d5c1a9af..b92dd7fd7 100644
--- a/gns3server/db/repositories/templates.py
+++ b/gns3server/db/repositories/templates.py
@@ -31,6 +31,8 @@ from .base import BaseRepository
import gns3server.db.models as models
from gns3server import schemas
+from gns3server.utils.image_inventory import image_lock
+from gns3server.controller.controller_error import ControllerNotFoundError
log = logging.getLogger(__name__)
@@ -136,18 +138,22 @@ class TemplatesRepository(BaseRepository):
await self._db_session.refresh(db_template)
return db_template
- async def get_image(self, image_path: str) -> Optional[models.Image]:
+ async def get_image(self, image_path: str, *, include_unavailable: bool = False) -> Optional[models.Image]:
"""
Get an image by its path.
"""
image_dir, image_name = os.path.split(image_path)
- if image_dir:
+ 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(image_path)
+ models.Image.filename == image_name, models.Image.path.endswith(os.sep + image_path, autoescape=True)
)
else:
query = select(models.Image).where(models.Image.filename == image_name)
+ 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()
@@ -164,20 +170,28 @@ class TemplatesRepository(BaseRepository):
Add an image to template.
"""
- query = (
- select(models.Template)
- .options(selectinload(models.Template.images))
- .where(models.Template.template_id == template_id)
- )
- result = await self._db_session.execute(query)
- template_in_db = result.scalars().first()
- if not template_in_db:
- return None
+ async with image_lock(image.path):
+ exists = (
+ await self._db_session.execute(
+ select(models.Image.image_id).where(models.Image.image_id == image.image_id)
+ )
+ ).scalar_one_or_none()
+ if exists is None:
+ raise ControllerNotFoundError(f"Image '{image.path}' was removed while creating the template")
+ query = (
+ select(models.Template)
+ .options(selectinload(models.Template.images))
+ .where(models.Template.template_id == template_id)
+ )
+ result = await self._db_session.execute(query)
+ template_in_db = result.scalars().first()
+ if not template_in_db:
+ return None
- template_in_db.images.append(image)
- await self._db_session.commit()
- await self._db_session.refresh(template_in_db)
- return template_in_db
+ template_in_db.images.append(image)
+ await self._db_session.commit()
+ await self._db_session.refresh(template_in_db)
+ return template_in_db
async def remove_image_from_template(self, template_id: UUID, image: models.Image) -> Union[None, models.Template]:
"""
diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py
index a9554c9cc..64f3f2673 100644
--- a/gns3server/db/tasks.py
+++ b/gns3server/db/tasks.py
@@ -15,8 +15,6 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-import asyncio
-import time
import os
from fastapi import FastAPI
@@ -31,13 +29,10 @@ from alembic import command, config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext
from alembic.util.exc import CommandError
-from watchdog.observers import Observer
-from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from gns3server.db.repositories.computes import ComputesRepository
from gns3server.db.repositories.images import ImagesRepository
-from gns3server.utils.images import md5sum, discover_images, read_image_info, InvalidImageError
-from gns3server.utils.asyncio import wait_run_in_executor
+from gns3server.utils.images import read_image_info
from gns3server import schemas
from .models import Base
@@ -142,7 +137,18 @@ async def connect_to_db(app: FastAPI) -> None:
log.info("Created new database and stamped to head revision")
elif db_state == "new_with_llm_configs":
# Database already has llm_model_configs table (from Base.metadata.create_all)
- # Just stamp the version
+ # Older unversioned metadata databases lack inventory fields.
+ # Ensure the additive schema before stamping the current head.
+ def upgrade_inventory(connection):
+ from alembic.operations import Operations
+ from gns3server.db_migrations.versions.d9e8a2b7c401_image_inventory_reconciliation import (
+ upgrade,
+ )
+
+ with Operations.context(MigrationContext.configure(connection)):
+ upgrade()
+
+ await conn.run_sync(upgrade_inventory)
await conn.run_sync(run_stamp, alembic_cfg)
await conn.commit()
log.info("Database has llm_model_configs table, stamped to head revision")
@@ -186,145 +192,25 @@ async def get_computes(app: FastAPI) -> List[schemas.Compute]:
return computes
-async def discover_images_on_filesystem(app: FastAPI) -> None:
-
- async with AsyncSession(app.state._db_engine) as db_session:
- images_repository = ImagesRepository(db_session)
- db_images = await images_repository.get_images()
- existing_image_paths = []
- for db_image in db_images:
- try:
- image = schemas.Image.model_validate(db_image)
- existing_image_paths.append(image.path)
- except ValidationError as e:
- log.error(f"Could not load image '{db_image.filename}' from database: {e}")
- continue
- for image_type in ("qemu", "ios", "iou"):
- discovered_images = await discover_images(image_type, existing_image_paths)
- for image_info in discovered_images:
- log.info(f"Adding discovered image '{image_info['path']}' to the database")
- try:
- await images_repository.add_image(**image_info)
- except SQLAlchemyError as e:
- log.warning(f"Error while adding image '{image_info['path']}' to the database: {e}")
-
- # monitor if images have been manually added
- asyncio.create_task(monitor_images_on_filesystem(app))
-
-
async def update_disk_checksums(updated_disks: List[str]) -> None:
- """
- Update the checksum of a list of disks in the database.
-
- :param updated_disks: list of updated disks
- """
-
+ """Refresh complete metadata after a server-managed disk modification."""
from gns3server.api.server import app
+ from gns3server.utils.image_inventory import image_lock
- async with AsyncSession(app.state._db_engine) as db_session:
- images_repository = ImagesRepository(db_session)
- for path in updated_disks:
- image = await images_repository.get_image(path)
- if image:
- log.info(f"Updating image '{path}' in the database")
- checksum = await wait_run_in_executor(md5sum, path, cache_to_md5file=False)
- if image.checksum != checksum:
- await images_repository.update_image(path, checksum, "md5")
-
-
-class EventHandler(PatternMatchingEventHandler):
- """
- Watchdog event handler.
- """
-
- def __init__(self, queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, **kwargs):
-
- self._loop = loop
- self._queue = queue
-
- # ignore temporary files, md5sum files, hidden files and directories
- super().__init__(ignore_patterns=["*.tmp", "*.md5sum", ".*"], ignore_directories=True, **kwargs)
-
- def on_closed(self, event: FileSystemEvent) -> None:
- # monitor for closed files (e.g. when a file has finished to be copied)
- if "/lib/" in event.src_path or "/lib64/" in event.src_path:
- return # ignore custom IOU libraries
- self._loop.call_soon_threadsafe(self._queue.put_nowait, event)
-
-
-class EventIterator(object):
- """
- Watchdog Event iterator.
- """
-
- def __init__(self, queue: asyncio.Queue):
- self.queue = queue
-
- def __aiter__(self):
- return self
-
- async def __anext__(self):
-
- item = await self.queue.get()
- if item is None:
- raise StopAsyncIteration
- return item
-
-
-async def monitor_images_on_filesystem(app: FastAPI):
-
- def watchdog(
- path: str, queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, app: FastAPI, recursive: bool = False
- ) -> None:
- """
- Thread to monitor a directory for new images.
- """
-
- handler = EventHandler(queue, loop)
- observer = Observer()
- observer.schedule(handler, str(path), recursive=recursive)
- observer.start()
- log.info(f"Monitoring for new images in '{path}'")
- while True:
- time.sleep(1)
- # stop when the app is exiting
- if app.state.exiting:
- observer.stop()
- observer.join(10)
- log.info(f"Stopping monitoring for new images in '{path}'")
- loop.call_soon_threadsafe(queue.put_nowait, None)
- break
-
- queue: asyncio.Queue = asyncio.Queue()
- loop = asyncio.get_event_loop()
- server_config = Config.instance().settings.Server
- image_dir = os.path.expanduser(server_config.images_path)
- asyncio.get_event_loop().run_in_executor(None, watchdog, image_dir, queue, loop, app, True)
-
- async for filesystem_event in EventIterator(queue):
- # read the file system event from the queue
- image_path = filesystem_event.src_path
- expected_image_type = None
- if "IOU" in image_path:
- expected_image_type = "iou"
- elif "QEMU" in image_path:
- expected_image_type = "qemu"
- elif "IOS" in image_path:
- expected_image_type = "ios"
- async with AsyncSession(app.state._db_engine) as db_session:
- images_repository = ImagesRepository(db_session)
- try:
- image = await read_image_info(image_path, expected_image_type)
- except InvalidImageError as e:
- log.warning(str(e))
- continue
- try:
- if await images_repository.get_image(image_path):
- continue
- await images_repository.add_image(**image)
- log.info(f"Discovered image '{image_path}' has been added to the database")
- except SQLAlchemyError as e:
- log.warning(f"Error while adding image '{image_path}' to the database: {e}")
+ for path in updated_disks:
+ async with image_lock(path):
+ async with AsyncSession(app.state._db_engine, expire_on_commit=False) as db_session:
+ repository = ImagesRepository(db_session)
+ image = await repository.get_image(path)
+ if image:
+ info = await read_image_info(path, str(image.image_type), allow_raw_image=True)
+ try:
+ os.unlink(path + ".md5sum")
+ except FileNotFoundError:
+ pass
+ except OSError as e:
+ log.warning("Could not invalidate checksum cache for '%s': %s", path, e)
+ await repository.save_verified_image(info)
async def get_user_llm_config_full(user_id: str, app: FastAPI) -> Optional[dict]:
diff --git a/gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py b/gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py
new file mode 100644
index 000000000..bb2ae487b
--- /dev/null
+++ b/gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py
@@ -0,0 +1,51 @@
+"""Track image availability and persistent synchronization jobs.
+
+Revision ID: d9e8a2b7c401
+Revises: c7e4a9f1d2b6
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "d9e8a2b7c401"
+down_revision = "c7e4a9f1d2b6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ # Startup also uses this for unversioned databases previously created from
+ # metadata, which may already contain some/all of these additive fields.
+ inspector = sa.inspect(op.get_bind())
+ existing = {column["name"] for column in inspector.get_columns("images")}
+ for column in (
+ sa.Column("availability", sa.String(), nullable=False, server_default="unknown"),
+ sa.Column("file_fingerprint", sa.String()),
+ sa.Column("last_seen_at", sa.DateTime()),
+ sa.Column("last_verified_at", sa.DateTime()),
+ sa.Column("last_error", sa.String()),
+ ):
+ if column.name not in existing:
+ op.add_column("images", column)
+ if inspector.has_table("image_sync_jobs"):
+ return
+ op.create_table(
+ "image_sync_jobs",
+ sa.Column("job_id", sa.String(), primary_key=True),
+ sa.Column("status", sa.String(), nullable=False),
+ sa.Column("dry_run", sa.Boolean(), nullable=False),
+ sa.Column("force_checksum", sa.Boolean(), nullable=False),
+ sa.Column("finished_at", sa.DateTime()),
+ sa.Column("counts", sa.JSON(), nullable=False),
+ sa.Column("errors", sa.JSON(), nullable=False),
+ sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp()),
+ sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp()),
+ )
+
+
+def downgrade():
+ op.drop_table("image_sync_jobs")
+ # Do not rebuild images: dropping the old table during a SQLite batch
+ # rebuild would cascade-delete image_template_map entries with FK enabled.
+ for name in ("last_error", "last_verified_at", "last_seen_at", "file_fingerprint", "availability"):
+ op.drop_column("images", name)
diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py
index 23056d58c..81e5fbc7e 100644
--- a/gns3server/schemas/__init__.py
+++ b/gns3server/schemas/__init__.py
@@ -41,7 +41,7 @@ from .controller.computes import (
Compute,
)
from .controller.templates import TemplateCreate, TemplateUpdate, TemplateUsage, Template
-from .controller.images import Image, ImageType
+from .controller.images import Image, ImageType, ImageSyncRequest, ImageSyncJob
from .controller.appliances import ApplianceVersion, ApplianceVersionV8, Appliance
from .controller.drawings import Drawing
from .controller.gns3vm import GNS3VM
diff --git a/gns3server/schemas/config.py b/gns3server/schemas/config.py
index 7b1eb19fb..c5becb3b5 100644
--- a/gns3server/schemas/config.py
+++ b/gns3server/schemas/config.py
@@ -184,6 +184,7 @@ class ServerSettings(BaseModel):
)
allow_raw_images: bool = Field(True, description="Allow raw images to be uploaded to the server")
auto_discover_images: bool = Field(True, description="Automatically discover images in the images directory")
+ image_sync_interval: int = Field(900, ge=10, description="Seconds between automatic image inventory scans")
report_errors: bool = Field(True, description="Automatically send crash reports to the GNS3 team")
additional_images_paths: List[str] = Field(
default_factory=list,
diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py
index ae98bcb6b..d2569fd26 100644
--- a/gns3server/schemas/controller/images.py
+++ b/gns3server/schemas/controller/images.py
@@ -16,6 +16,8 @@
from pydantic import ConfigDict, BaseModel, Field
from enum import Enum
+from datetime import datetime
+from typing import Optional, Literal
from .base import DateTimeModelMixin
@@ -37,7 +39,28 @@ class ImageBase(BaseModel):
image_size: int = Field(..., description="Image size in bytes")
checksum: str = Field(..., description="Checksum value")
checksum_algorithm: str = Field(..., description="Checksum algorithm")
+ availability: Literal["unknown", "available", "missing", "unavailable", "invalid"] = "unknown"
+ last_seen_at: Optional[datetime] = None
+ last_verified_at: Optional[datetime] = None
+ last_error: Optional[str] = None
class Image(DateTimeModelMixin, ImageBase):
model_config = ConfigDict(from_attributes=True)
+
+
+class ImageSyncRequest(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+ dry_run: bool = False
+ force_checksum: bool = False
+
+
+class ImageSyncJob(DateTimeModelMixin):
+ model_config = ConfigDict(from_attributes=True)
+ job_id: str
+ status: Literal["queued", "running", "completed", "partial", "failed", "cancelled", "interrupted"]
+ dry_run: bool
+ force_checksum: bool
+ finished_at: Optional[datetime] = None
+ counts: dict[str, int]
+ errors: list[dict[str, str]]
diff --git a/gns3server/schemas/controller/settings.py b/gns3server/schemas/controller/settings.py
index 9925c403c..87b4ab2a5 100644
--- a/gns3server/schemas/controller/settings.py
+++ b/gns3server/schemas/controller/settings.py
@@ -120,6 +120,7 @@ class ServerSettingsUpdate(BaseModel):
default_symbol_theme: Optional[BuiltinSymbolTheme] = None
allow_raw_images: Optional[bool] = None
auto_discover_images: Optional[bool] = None
+ image_sync_interval: Optional[int] = Field(None, ge=10)
report_errors: Optional[bool] = None
additional_images_paths: Optional[List[str]] = None
console_start_port_range: Optional[int] = Field(None, gt=0, le=65535)
diff --git a/gns3server/services/image_reconciliation.py b/gns3server/services/image_reconciliation.py
new file mode 100644
index 000000000..2fa523872
--- /dev/null
+++ b/gns3server/services/image_reconciliation.py
@@ -0,0 +1,510 @@
+"""Recoverable controller image inventory, shared by manual and automatic sync."""
+
+import asyncio
+import logging
+import os
+import stat
+import threading
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import select, update, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+from watchdog.events import FileSystemEventHandler
+from watchdog.observers import Observer
+
+from gns3server.config import Config
+from gns3server.db.models import Image, ImageSyncJob
+from gns3server.db.repositories.images import ImagesRepository
+from gns3server.utils.image_inventory import (
+ ImageLock,
+ ImageLockBusy,
+ contained_path,
+ fingerprint,
+ image_lock,
+ normalized_path,
+)
+from gns3server.utils.images import inspect_image_file, InvalidImageError, ImageChangedError
+
+log = logging.getLogger(__name__)
+COUNTERS = (
+ "scanned",
+ "added",
+ "updated",
+ "unchanged",
+ "missing",
+ "unavailable",
+ "invalid",
+ "deferred",
+ "out_of_scope",
+ "errors",
+ "bytes_hashed",
+)
+RAW_EXTENSIONS = {".raw", ".img", ".iso", ".fd", ".vhd", ".vdi", ".bin"}
+
+
+def utcnow():
+ return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+def ignored(name):
+ return name.startswith(".") or name.endswith((".tmp", ".md5sum"))
+
+
+def configured_roots():
+ settings = Config.instance().settings.Server
+ roots = sorted({normalized_path(p) for p in [settings.images_path, *settings.additional_images_paths]})
+ # Keep parent roots only, avoiding duplicate scans and conflicting absence passes.
+ return [root for root in roots if not any(root != other and contained_path(root, other) for other in roots)]
+
+
+def root_identity(root):
+ info = os.stat(root)
+ # Opening the directory tests readability, not just existence.
+ with os.scandir(root):
+ pass
+ return info.st_dev, info.st_ino, os.path.realpath(root)
+
+
+def enumerate_root(root):
+ """Observe a root without creating directories or silently dropping failures."""
+ files, errors = {}, []
+ try:
+ root_stat = os.stat(root)
+ if not stat.S_ISDIR(root_stat.st_mode):
+ raise NotADirectoryError(root)
+ for directory, dirs, names in os.walk(
+ root,
+ onerror=lambda e: errors.append({"path": str(e.filename or root), "reason": str(e)}),
+ followlinks=False,
+ ):
+ dirs[:] = [
+ name
+ for name in dirs
+ if not ignored(name)
+ and name not in ("lib", "lib64")
+ and not os.path.islink(os.path.join(directory, name))
+ ]
+ for name in names:
+ if ignored(name):
+ continue
+ path = normalized_path(os.path.join(directory, name))
+ try:
+ files[path] = fingerprint(path)
+ except OSError as e:
+ errors.append({"path": path, "reason": str(e)})
+ # A mount/root replacement during traversal makes the absence pass unsafe.
+ after = os.stat(root)
+ if (root_stat.st_dev, root_stat.st_ino) != (after.st_dev, after.st_ino):
+ errors.append({"path": root, "reason": "Image root changed during scan"})
+ except OSError as e:
+ errors.append({"path": root, "reason": str(e)})
+ return files, errors
+
+
+class InventoryEvents(FileSystemEventHandler):
+ def __init__(self, loop, dirty):
+ self.loop, self.dirty = loop, dirty
+ self._pending = False
+ self._lock = threading.Lock()
+
+ def _mark_dirty(self):
+ with self._lock:
+ self._pending = False
+ self.dirty.set()
+
+ def on_any_event(self, event):
+ if event.event_type not in ("created", "modified", "closed", "deleted", "moved"):
+ return
+ paths = [event.src_path, getattr(event, "dest_path", "")]
+ if any(path and not ignored(os.path.basename(path)) for path in paths):
+ # A single dirty bit coalesces events with bounded memory, including
+ # directory renames and bursts. The next scan observes current state.
+ with self._lock:
+ if self._pending:
+ return
+ self._pending = True
+ self.loop.call_soon_threadsafe(self._mark_dirty)
+
+
+class ImageReconciliationService:
+ def __init__(self, engine, settle_seconds=1.0):
+ self.engine = engine
+ self.settle_seconds = settle_seconds
+ self.task = None
+ self.scheduler = None
+ self.observer = None
+ self.dirty = asyncio.Event()
+ self.stopping = threading.Event()
+ self._watched_roots = None
+ self._catalog_paths = {}
+ self._ambiguous_paths = set()
+
+ async def start(self, dry_run=False, force_checksum=False):
+ if self.stopping.is_set():
+ raise ImageLockBusy("Image synchronization is shutting down")
+ lock = ImageLock("image-inventory", wait=False)
+ await lock.__aenter__()
+ try:
+ async with AsyncSession(self.engine, expire_on_commit=False) as session:
+ # The OS lock proves no previous worker using this catalog is
+ # active. Recover jobs abandoned by a crash before admitting one.
+ await session.execute(
+ update(ImageSyncJob)
+ .where(ImageSyncJob.status.in_(["queued", "running"]))
+ .values(status="interrupted", finished_at=utcnow())
+ )
+ job = ImageSyncJob(
+ job_id=str(uuid.uuid4()),
+ status="queued",
+ dry_run=dry_run,
+ force_checksum=force_checksum,
+ counts=dict.fromkeys(COUNTERS, 0),
+ errors=[],
+ )
+ session.add(job)
+ # Bound history; active jobs are never removed.
+ old_jobs = (
+ select(ImageSyncJob.job_id)
+ .where(ImageSyncJob.status.notin_(["queued", "running"]))
+ .order_by(ImageSyncJob.created_at.desc())
+ .offset(99)
+ )
+ await session.execute(delete(ImageSyncJob).where(ImageSyncJob.job_id.in_(old_jobs)))
+ await session.commit()
+ await session.refresh(job)
+ result = job.asdict()
+ self.task = asyncio.create_task(self._run(result, lock), name="image-inventory-sync")
+ self.task.add_done_callback(self._finished)
+ return result
+ except BaseException:
+ await lock.__aexit__()
+ raise
+
+ @staticmethod
+ def _finished(task):
+ if not task.cancelled() and task.exception() is not None:
+ log.error("Could not persist image synchronization outcome: %s", task.exception())
+
+ async def get_job(self, job_id, offset=0, limit=100):
+ # A client can resume polling after a restart even with automatic scans
+ # disabled. Do not leave an abandoned job looking permanently active.
+ try:
+ async with ImageLock("image-inventory", wait=False):
+ async with AsyncSession(self.engine) as session:
+ await session.execute(
+ update(ImageSyncJob)
+ .where(ImageSyncJob.status.in_(["queued", "running"]))
+ .values(status="interrupted", finished_at=utcnow())
+ )
+ await session.commit()
+ except ImageLockBusy:
+ pass
+ async with AsyncSession(self.engine) as session:
+ job = await session.get(ImageSyncJob, job_id)
+ if job is None:
+ return None
+ result = job.asdict()
+ result["errors"] = result["errors"][offset : offset + limit]
+ return result
+
+ async def _persist(self, job):
+ async with AsyncSession(self.engine) as session:
+ await session.execute(
+ update(ImageSyncJob)
+ .where(ImageSyncJob.job_id == job["job_id"])
+ .values(
+ status=job["status"],
+ counts=dict(job["counts"]),
+ errors=list(job["errors"]),
+ finished_at=job.get("finished_at"),
+ )
+ )
+ await session.commit()
+
+ def _error(self, job, path, reason):
+ job["counts"]["errors"] += 1
+ # The total count remains accurate even when detailed errors are capped.
+ if len(job["errors"]) < 1000:
+ job["errors"].append({"path": path, "reason": str(reason)})
+
+ async def _inspect(self, path, expected, allow_raw):
+ worker = asyncio.create_task(asyncio.to_thread(inspect_image_file, path, expected, allow_raw, self.stopping))
+ try:
+ return await asyncio.shield(worker)
+ except asyncio.CancelledError:
+ self.stopping.set()
+ try:
+ await worker
+ except (OSError, InvalidImageError):
+ pass
+ raise
+
+ async def _state(self, row, availability, reason, dry_run):
+ if row is None or dry_run:
+ return
+ async with AsyncSession(self.engine) as session:
+ # Revision guard for legacy writers not yet taking the path lock.
+ await session.execute(
+ update(Image)
+ .where(
+ Image.image_id == row["image_id"],
+ Image.path == row["path"],
+ Image.file_fingerprint == row["file_fingerprint"],
+ Image.checksum == row["checksum"],
+ )
+ .values(availability=availability, last_error=str(reason))
+ )
+ await session.commit()
+
+ async def _row(self, path):
+ async with AsyncSession(self.engine) as session:
+ stored = self._catalog_paths.get(normalized_path(path), path)
+ row = (await session.execute(select(Image).where(Image.path.in_([stored, path])))).scalar_one_or_none()
+ return row.asdict() if row else None
+
+ async def _file(self, path, observed, job, root): # noqa: C901 - keep per-file outcomes under one path lock
+ if path in self._ambiguous_paths:
+ job["counts"]["deferred"] += 1
+ return
+ async with image_lock(path):
+ row = await self._row(path)
+ counts = job["counts"]
+ counts["scanned"] += 1
+ try:
+ if not contained_path(os.path.realpath(path), root):
+ raise OSError("Image path resolves outside its configured root")
+ current = await asyncio.to_thread(fingerprint, path)
+ if current != observed:
+ raise ImageChangedError("File changed during settling interval")
+ if (
+ row
+ and row["availability"] == "available"
+ and row["file_fingerprint"] == current
+ and not job["force_checksum"]
+ ):
+ counts["unchanged"] += 1
+ return
+ expected = row["image_type"] if row else None
+ main_root = normalized_path(Config.instance().settings.Server.images_path)
+ if not expected and contained_path(path, main_root):
+ component = os.path.relpath(path, main_root).split(os.sep)[0]
+ expected = {"QEMU": "qemu", "IOS": "ios", "IOU": "iou"}.get(component)
+ allow_raw = Config.instance().settings.Server.allow_raw_images and (
+ expected == "qemu"
+ or (expected in (None, "qemu") and os.path.splitext(path)[1].lower() in RAW_EXTENSIONS)
+ )
+ info = await self._inspect(path, expected, allow_raw)
+ if row:
+ # Preserve legacy path spelling as well as image identity.
+ info["path"] = row["path"]
+ counts["bytes_hashed"] += info["image_size"]
+ if self.stopping.is_set():
+ raise asyncio.CancelledError
+ if await asyncio.to_thread(fingerprint, path) != info["file_fingerprint"]:
+ raise ImageChangedError("File changed before catalog update")
+ if not job["dry_run"]:
+ # Do not leave stale sidecars for legacy compute consumers.
+ try:
+ os.unlink(path + ".md5sum")
+ except FileNotFoundError:
+ pass
+ except OSError as e:
+ self._error(job, path + ".md5sum", e)
+ async with AsyncSession(self.engine, expire_on_commit=False) as session:
+ await ImagesRepository(session).save_verified_image(info)
+ counts["updated" if row else "added"] += 1
+ except ImageChangedError as e:
+ counts["deferred"] += 1
+ await self._state(row, "unavailable", e, job["dry_run"])
+ except InvalidImageError as e:
+ counts["invalid"] += 1
+ await self._state(row, "invalid", e, job["dry_run"])
+ self._error(job, path, e)
+ except OSError as e:
+ if self.stopping.is_set():
+ raise asyncio.CancelledError from e
+ counts["unavailable"] += 1
+ await self._state(row, "unavailable", e, job["dry_run"])
+ self._error(job, path, e)
+
+ async def _run(self, job, lock): # noqa: C901 - one job owns root health, progress and lock lifetime
+ job["status"] = "running"
+ try:
+ await self._persist(job)
+ async with AsyncSession(self.engine) as session:
+ paths = (await session.execute(select(Image.path))).scalars().all()
+ self._catalog_paths = {}
+ self._ambiguous_paths = set()
+ roots = configured_roots()
+ for path in paths:
+ normalized = normalized_path(path)
+ if not any(contained_path(path, root) for root in roots):
+ job["counts"]["out_of_scope"] += 1
+ self._error(job, path, "Catalog path is outside the configured image directories; record retained")
+ if normalized in self._catalog_paths:
+ self._ambiguous_paths.add(normalized)
+ self._error(job, path, "Multiple catalog paths resolve to the same image; manual review required")
+ self._catalog_paths[normalized] = path
+ for root in roots:
+ if self.stopping.is_set():
+ raise asyncio.CancelledError
+ resolved_root = os.path.realpath(root)
+ try:
+ identity = await asyncio.to_thread(root_identity, root)
+ except OSError:
+ identity = None
+ files, errors = await asyncio.to_thread(enumerate_root, root)
+ if resolved_root != os.path.realpath(root):
+ errors.append({"path": root, "reason": "Image root target changed during scan"})
+ for error in errors:
+ self._error(job, error["path"], error["reason"])
+ # One settling wait per root, not one per large image collection.
+ if files and self.settle_seconds:
+ await asyncio.sleep(self.settle_seconds)
+ for path, observed in files.items():
+ if self.stopping.is_set():
+ raise asyncio.CancelledError
+ try:
+ await self._file(path, observed, job, resolved_root)
+ except Exception as e:
+ self._error(job, path, e)
+ log.warning("Could not reconcile image %s: %s", path, e)
+ if job["counts"]["scanned"] % 25 == 0:
+ await self._persist(job)
+ try:
+ if identity != await asyncio.to_thread(root_identity, root):
+ raise OSError("Image root changed during reconciliation")
+ except OSError as e:
+ errors.append({"path": root, "reason": str(e)})
+ self._error(job, root, e)
+ async with AsyncSession(self.engine) as session:
+ rows = [image.asdict() for image in (await session.execute(select(Image))).scalars()]
+ for row in rows:
+ if self.stopping.is_set():
+ raise asyncio.CancelledError
+ path = row["path"]
+ if (
+ not contained_path(path, root)
+ or normalized_path(path) in files
+ or normalized_path(path) in self._ambiguous_paths
+ ):
+ continue
+ async with image_lock(path):
+ row = await self._row(path)
+ if row is None:
+ continue
+ if errors:
+ job["counts"]["unavailable"] += 1
+ await self._state(
+ row, "unavailable", "Image root was not completely scanned", job["dry_run"]
+ )
+ continue
+ try:
+ if not contained_path(os.path.realpath(path), resolved_root):
+ raise OSError("Image path resolves outside its configured root")
+ await asyncio.to_thread(fingerprint, path)
+ except FileNotFoundError:
+ try:
+ if identity != await asyncio.to_thread(root_identity, root):
+ raise OSError("Image root changed during reconciliation")
+ except OSError as e:
+ job["counts"]["unavailable"] += 1
+ await self._state(row, "unavailable", e, job["dry_run"])
+ self._error(job, root, e)
+ else:
+ job["counts"]["missing"] += 1
+ await self._state(row, "missing", "Image file is missing", job["dry_run"])
+ except OSError as e:
+ job["counts"]["unavailable"] += 1
+ await self._state(row, "unavailable", e, job["dry_run"])
+ self._error(job, path, e)
+ if self.stopping.is_set():
+ raise asyncio.CancelledError
+ job["status"] = "partial" if job["counts"]["errors"] or job["counts"]["deferred"] else "completed"
+ except asyncio.CancelledError:
+ job["status"] = "cancelled"
+ except Exception as e:
+ job["status"] = "failed"
+ self._error(job, "", e)
+ log.exception("Image reconciliation failed")
+ finally:
+ job["finished_at"] = utcnow()
+ try:
+ await self._persist(job)
+ finally:
+ await lock.__aexit__()
+
+ async def _watch(self):
+ roots = configured_roots() if Config.instance().settings.Server.auto_discover_images else []
+ # Include availability in the signature so newly mounted roots get watched.
+ roots = [root for root in roots if os.path.isdir(root)]
+ if roots == self._watched_roots:
+ return
+ if self.observer:
+ self.observer.stop()
+ await asyncio.to_thread(self.observer.join)
+ self.observer = None
+ self._watched_roots = roots
+ if roots:
+ observer = Observer()
+ handler = InventoryEvents(asyncio.get_running_loop(), self.dirty)
+ try:
+ for root in roots:
+ observer.schedule(handler, root, recursive=True)
+ observer.start()
+ self.observer = observer
+ except OSError:
+ observer.stop()
+ if observer.is_alive():
+ await asyncio.to_thread(observer.join)
+ self._watched_roots = None
+ log.warning("Image watcher unavailable; periodic scanning remains active", exc_info=True)
+
+ async def _automatic(self):
+ await asyncio.sleep(5)
+ while not self.stopping.is_set():
+ try:
+ await self._watch()
+ self.dirty.clear()
+ if Config.instance().settings.Server.auto_discover_images:
+ try:
+ await self.start()
+ await asyncio.shield(self.task)
+ except ImageLockBusy:
+ pass
+ try:
+ await asyncio.wait_for(self.dirty.wait(), Config.instance().settings.Server.image_sync_interval)
+ await asyncio.sleep(1) # debounce bursts of file/directory events
+ except asyncio.TimeoutError:
+ pass
+ except asyncio.CancelledError:
+ break
+ except Exception:
+ log.exception("Automatic image synchronization failed; will retry")
+ await asyncio.sleep(10)
+
+ def start_background(self):
+ self.scheduler = asyncio.create_task(self._automatic(), name="image-inventory-scheduler")
+
+ async def close(self):
+ self.stopping.set()
+ if self.scheduler:
+ self.scheduler.cancel()
+ await asyncio.gather(self.scheduler, return_exceptions=True)
+ if self.observer:
+ self.observer.stop()
+ await asyncio.to_thread(self.observer.join)
+ self.observer = None
+ if self.task and not self.task.done():
+ # Cooperative cancellation also covers a task not yet entered; its
+ # finally block must run to release the already-acquired job lock.
+ await asyncio.gather(self.task, return_exceptions=True)
+
+
+def get_image_reconciliation_service(app):
+ service = getattr(app.state, "image_reconciliation", None)
+ if service is None:
+ service = ImageReconciliationService(app.state._db_engine)
+ app.state.image_reconciliation = service
+ return service
diff --git a/gns3server/services/templates.py b/gns3server/services/templates.py
index 00f02b4b0..c4fd274ed 100644
--- a/gns3server/services/templates.py
+++ b/gns3server/services/templates.py
@@ -279,7 +279,8 @@ class TemplatesService:
if not image_path:
return
- image = await self._templates_repo.get_image(image_path)
+ # Removing an association must also find missing/invalid image rows.
+ image = await self._templates_repo.get_image(image_path, include_unavailable=True)
if image is None:
return
await self._templates_repo.remove_image_from_template(template_id, image)
diff --git a/gns3server/utils/image_inventory.py b/gns3server/utils/image_inventory.py
new file mode 100644
index 000000000..78943ba11
--- /dev/null
+++ b/gns3server/utils/image_inventory.py
@@ -0,0 +1,99 @@
+"""Filesystem primitives shared by image writers and inventory reconciliation."""
+
+import asyncio
+import hashlib
+import os
+import stat
+
+from gns3server.config import Config
+
+
+def normalized_path(path):
+ return os.path.normcase(os.path.abspath(os.path.expanduser(path)))
+
+
+def contained_path(path, root):
+ try:
+ return os.path.commonpath((normalized_path(path), normalized_path(root))) == normalized_path(root)
+ except ValueError:
+ return False
+
+
+def fingerprint(path):
+ info = os.stat(path, follow_symlinks=True)
+ if not stat.S_ISREG(info.st_mode):
+ raise OSError(f"Not a regular image file: {path}")
+ return stat_fingerprint(info)
+
+
+def stat_fingerprint(info):
+ # Store as text: inode/device numbers need not fit a signed SQL BIGINT.
+ return ":".join(
+ str(value) for value in (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
+ )
+
+
+class ImageLockBusy(Exception):
+ pass
+
+
+class ImageLock:
+ """Advisory lock shared by controller processes using the same config directory.
+
+ Lock files are deliberately retained: unlinking them permits two processes to
+ lock different inodes for the same name. OS locks are released on process exit.
+ """
+
+ def __init__(self, key, wait=True):
+ self.key = key
+ self.wait = wait
+ self._file = None
+
+ async def __aenter__(self):
+ directory = os.path.join(Config.instance().config_dir, ".image-locks")
+ os.makedirs(directory, exist_ok=True)
+ path = os.path.join(directory, hashlib.sha256(self.key.encode()).hexdigest() + ".lock")
+ self._file = open(path, "a+b")
+ if os.name == "nt" and os.fstat(self._file.fileno()).st_size == 0:
+ self._file.write(b"\0")
+ self._file.flush()
+ try:
+ while True:
+ try:
+ if os.name == "nt":
+ import msvcrt
+
+ self._file.seek(0)
+ msvcrt.locking(self._file.fileno(), msvcrt.LK_NBLCK, 1)
+ else:
+ import fcntl
+
+ fcntl.flock(self._file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ return self
+ except (BlockingIOError, PermissionError):
+ if not self.wait:
+ raise ImageLockBusy(self.key)
+ await asyncio.sleep(0.05)
+ except BaseException:
+ self._file.close()
+ self._file = None
+ raise
+
+ async def __aexit__(self, *args):
+ if self._file is not None:
+ self._file.close()
+ self._file = None
+
+
+def image_lock(path):
+ return ImageLock("image:" + normalized_path(os.path.realpath(path)))
+
+
+def publish_image(temporary, destination):
+ """Atomically publish a complete file without replacing an existing image.
+
+ Both paths must be on the same filesystem. Hard linking provides the atomic
+ no-overwrite operation that os.replace()/shutil.move() cannot provide.
+ """
+ os.link(temporary, destination)
+ os.unlink(temporary)
diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py
index 06b4ae9be..e2c458e1b 100644
--- a/gns3server/utils/images.py
+++ b/gns3server/utils/images.py
@@ -18,7 +18,10 @@ import os
import hashlib
import stat
import aiofiles
-import shutil
+import asyncio
+import tempfile
+
+from gns3server.utils.image_inventory import fingerprint, stat_fingerprint, image_lock, publish_image, contained_path
try:
import importlib_resources
@@ -126,33 +129,46 @@ def get_builtin_disks() -> List[str]:
return builtin_disks
-async def read_image_info(path: str, expected_image_type: str | None = None) -> dict:
+def inspect_image_file(path, expected_image_type=None, allow_raw_image=False, stopped_event=None):
+ """Read a stable regular file once, never trusting checksum sidecars."""
+ before = fingerprint(path)
+ flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
+ with os.fdopen(os.open(path, flags), "rb") as f:
+ 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)
+ 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)
+ if expected_image_type and image_type != expected_image_type:
+ raise InvalidImageError(f"Detected image type for '{path}' is {image_type}, expected {expected_image_type}")
+ digest = hashlib.md5(header)
+ while True:
+ if stopped_event is not None and stopped_event.is_set():
+ raise InterruptedError("Image inspection cancelled")
+ chunk = f.read(1024 * 1024)
+ if not chunk:
+ break
+ digest.update(chunk)
+ if stat_fingerprint(os.fstat(f.fileno())) != before or fingerprint(path) != before:
+ raise ImageChangedError(f"Image changed while reading: {path}")
+ return dict(
+ image_name=os.path.basename(path),
+ image_type=image_type,
+ image_size=info.st_size,
+ path=path,
+ checksum=digest.hexdigest(),
+ checksum_algorithm="md5",
+ file_fingerprint=before,
+ )
- header_magic_len = 7
+
+async def read_image_info(path: str, expected_image_type: str | None = None, allow_raw_image=False) -> dict:
try:
- async with aiofiles.open(path, "rb") as f:
- image_header = await f.read(header_magic_len) # read the first 7 bytes of the file
- if len(image_header) >= header_magic_len:
- detected_image_type = check_valid_image_header(path, image_header)
- if expected_image_type and detected_image_type != expected_image_type:
- raise InvalidImageError(
- f"Detected image type for '{path}' is {detected_image_type}, "
- f"expected type is {expected_image_type}"
- )
- else:
- raise InvalidImageError(f"Image '{path}' is too small to be valid")
+ return await asyncio.to_thread(inspect_image_file, path, expected_image_type, allow_raw_image)
except OSError as e:
- raise InvalidImageError(f"Cannot read image '{path}': {e}")
-
- image_info = {
- "image_name": os.path.basename(path),
- "image_type": detected_image_type,
- "image_size": os.stat(path).st_size,
- "path": path,
- "checksum": await wait_run_in_executor(md5sum, path, cache_to_md5file=False),
- "checksum_algorithm": "md5",
- }
- return image_info
+ raise InvalidImageError(f"Cannot read image '{path}': {e}") from e
async def discover_images(image_type: str, skip_image_paths: list | None = None) -> List[dict]:
@@ -244,7 +260,7 @@ def images_directories(image_type, include_parent_directory=True):
return [force_unix_path(p) for p in paths if os.path.exists(p)]
-def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
+def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True, use_cache=True):
"""
Return the md5sum of an image and cache it on disk
@@ -263,7 +279,7 @@ def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
else:
md5sum_file = path + ".md5sum"
- if os.path.exists(md5sum_file):
+ if use_cache and os.path.exists(md5sum_file):
try:
with open(md5sum_file) as f:
md5 = f.read().strip()
@@ -319,6 +335,10 @@ class InvalidImageError(Exception):
return self._message
+class ImageChangedError(InvalidImageError):
+ """An observation must be retried because the file is still changing."""
+
+
def check_valid_image_header(path: str, data: bytes, allow_raw_image: bool = False) -> str:
if data[:7] == b"\x7fELF\x01\x02\x01":
@@ -347,54 +367,66 @@ async def write_image(
) -> models.Image:
image_dir, image_name = os.path.split(image_filename)
- # Store the file under its final name only when the upload is completed
- tmp_path = image_path + ".tmp"
- log.info(f"Writing image file to '{tmp_path}'")
+ # HTTP chunk boundaries need not align with the seven-byte image header.
+ iterator = stream.__aiter__()
+ prefix = bytearray()
+ async for chunk in iterator:
+ prefix.extend(chunk)
+ if len(prefix) >= 7:
+ break
+ if len(prefix) < 7:
+ 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))
+ 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()
- header_magic_len = 7
- image_type = None
- image_size = 0
try:
async with aiofiles.open(tmp_path, "wb") as f:
- async for chunk in stream:
- if check_image_header and len(chunk) >= header_magic_len:
- check_image_header = False
- image_type = check_valid_image_header(image_path, chunk, allow_raw_image)
+ 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)
- if not image_size or image_size < header_magic_len:
- raise InvalidImageError("The image content is empty or too small to be valid")
-
- if not image_dir:
- directory = default_images_directory(image_type)
- os.makedirs(directory, exist_ok=True)
- image_path = os.path.abspath(os.path.join(directory, image_filename))
-
- if os.path.exists(image_path):
- raise InvalidImageError(
- f"File '{image_path}' already exists, please choose a different name or remove the existing image"
+ 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),
+ )
)
-
- checksum_hex = checksum.hexdigest()
- image_dir = os.path.dirname(image_path)
- duplicate_image = await images_repo.get_image_by_checksum(checksum_hex, image_dir)
- if duplicate_image:
- raise InvalidImageError(
- f"Image '{duplicate_image.filename}' with the same checksum already exists in '{image_dir}'"
- )
-
- shutil.move(tmp_path, image_path)
- os.chmod(image_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
+ if image is None:
+ raise InvalidImageError(f"Failed to save image '{image_name}' to database")
+ return image
finally:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except OSError:
log.warning(f"Could not remove '{tmp_path}'")
-
- return await images_repo.add_image(
- image_name, image_type, image_size, image_path, checksum_hex, checksum_algorithm="md5"
- )
diff --git a/pyproject.toml b/pyproject.toml
index f979b5eb3..07c46dc7a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -277,7 +277,7 @@ extend-immutable-calls = [
"tests/test_utils.py" = ["F403", "F405"]
"tests/utils/test_asyncio.py" = ["B017"]
"tests/utils/test_packet_filter_validation.py" = ["W292"]
-
+"tests/services/test_image_reconciliation.py" = ["S324"]
[tool.mypy]
files = ["gns3server"]
diff --git a/tests/api/routes/controller/test_image_sync.py b/tests/api/routes/controller/test_image_sync.py
new file mode 100644
index 000000000..6e1a43d42
--- /dev/null
+++ b/tests/api/routes/controller/test_image_sync.py
@@ -0,0 +1,138 @@
+import os
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+
+from gns3server.db.repositories.images import ImagesRepository
+from gns3server.services.image_reconciliation import ImageReconciliationService
+from gns3server.services import auth_service
+from gns3server.services.authentication import DEFAULT_JWT_SECRET_KEY
+
+pytestmark = pytest.mark.asyncio
+QCOW = b"QFI\xfb\x00\x00\x00"
+
+
+@pytest_asyncio.fixture
+async def sync_service(app, db_engine, db_session, monkeypatch):
+ service = ImageReconciliationService(db_engine, settle_seconds=0)
+ monkeypatch.setattr(app.state, "image_reconciliation", service, raising=False)
+ try:
+ yield service
+ finally:
+ await service.close()
+
+
+class TestImageSyncRoutes:
+ async def test_manual_sync_with_auto_discovery_disabled(self, app, client, sync_service, config, images_dir):
+ assert not config.settings.Server.auto_discover_images
+ path = Path(images_dir) / "QEMU" / "manual.qcow2"
+ path.write_bytes(QCOW)
+ response = await client.post("/v3/images/sync", json={})
+ assert response.status_code == 202
+ assert response.json()["job_id"] in response.headers["Location"]
+ await sync_service.task
+ job = await client.get(response.headers["Location"])
+ assert job.status_code == 200
+ assert job.json()["status"] == "completed"
+ assert job.json()["counts"]["added"] == 1
+ listed = await client.get("/v3/images", params={"availability": "available"})
+ assert any(image["path"] == str(path) for image in listed.json())
+ path.unlink()
+ await client.post("/v3/images/sync", json={})
+ await sync_service.task
+ missing = await client.get("/v3/images", params={"availability": "missing"})
+ assert any(image["path"] == str(path) for image in missing.json())
+
+ async def test_dry_run_and_request_validation(self, client, sync_service, images_dir):
+ path = Path(images_dir) / "QEMU" / "preview.qcow2"
+ path.write_bytes(QCOW)
+ response = await client.post("/v3/images/sync", json={"dry_run": True, "force_checksum": True})
+ assert response.status_code == 202
+ await sync_service.task
+ result = await client.get(response.headers["Location"], params={"offset": 0, "limit": 1})
+ assert result.json()["counts"]["added"] == 1
+ listed = await client.get("/v3/images")
+ assert all(image["path"] != str(path) for image in listed.json())
+ assert (await client.post("/v3/images/sync", json={"path": "/etc"})).status_code == 422
+ assert (await client.get(response.headers["Location"], params={"limit": 1001})).status_code == 422
+ assert (await client.get("/v3/images/sync/jobs/does-not-exist")).status_code == 404
+
+ async def test_overlapping_sync_is_conflict(self, client, sync_service):
+ from gns3server.utils.image_inventory import ImageLock
+
+ async with ImageLock("image-inventory"):
+ response = await client.post("/v3/images/sync", json={})
+ assert response.status_code == 409
+
+ async def test_delete_and_prune_keep_rows_on_permission_error(self, client, db_session, images_dir):
+ response = await client.post("/v3/images/upload/protected.qcow2", content=QCOW)
+ assert response.status_code == 201
+ path = response.json()["path"]
+ with patch("gns3server.api.routes.controller.images.os.remove", side_effect=PermissionError("denied")):
+ response = await client.delete("/v3/images/protected.qcow2")
+ assert response.status_code == 409
+ response = await client.delete("/v3/images/prune")
+ assert response.status_code == 409
+ assert os.path.exists(path)
+ assert await ImagesRepository(db_session).get_image(path) is not None
+ os.unlink(path)
+ assert (await client.delete("/v3/images/protected.qcow2")).status_code == 204
+
+ async def test_create_does_not_overwrite_unindexed_file(self, client, images_dir):
+ path = Path(images_dir) / "QEMU" / "existing.qcow2"
+ path.write_bytes(QCOW + b"preserve me")
+ response = await client.post("/v3/images/qemu/existing.qcow2", json={"format": "qcow2", "size": 1})
+ assert response.status_code == 400
+ assert path.read_bytes() == QCOW + b"preserve me"
+
+ async def test_upload_rejects_path_prefix_sibling_and_symlink(self, client, images_dir, tmp_path):
+ sibling = images_dir + "-outside/escape.qcow2"
+ response = await client.post("/v3/images/upload/" + sibling, content=QCOW)
+ assert response.status_code == 403
+ assert not os.path.exists(sibling)
+ link = Path(images_dir) / "external"
+ link.symlink_to(tmp_path, target_is_directory=True)
+ response = await client.post("/v3/images/upload/external/escape.qcow2", content=QCOW)
+ assert response.status_code == 403
+ assert not (tmp_path / "escape.qcow2").exists()
+
+ async def test_sync_requires_allocate_privilege(self, client, sync_service, test_user):
+ 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 client.post("/v3/images/sync", json={}, headers={"Authorization": f"Bearer {token}"})
+ assert response.status_code == 403
+ assert "Image.Allocate" in response.text
+ response = await client.get("/v3/images/sync/jobs/test", headers={"Authorization": f"Bearer {token}"})
+ assert response.status_code == 403
+ assert "Image.Audit" in response.text
+ response = await client.post("/v3/images/sync", json={}, headers={"Authorization": ""})
+ assert response.status_code == 401
+
+ async def test_template_can_detach_missing_image(self, app, client, sync_service, db_session):
+ uploaded = await client.post("/v3/images/upload/detach-missing.qcow2", content=QCOW)
+ assert uploaded.status_code == 201
+ image = uploaded.json()
+ image_id = (await ImagesRepository(db_session).get_image(image["path"])).image_id
+ created = await client.post(
+ "/v3/templates",
+ json={
+ "name": "Detach missing image",
+ "compute_id": "local",
+ "template_type": "qemu",
+ "hda_disk_image": image["filename"],
+ },
+ )
+ assert created.status_code == 201
+ template_id = created.json()["template_id"]
+ os.unlink(image["path"])
+ await client.post("/v3/images/sync", json={})
+ await sync_service.task
+ updated = await client.put(f"/v3/templates/{template_id}", json={"hda_disk_image": ""})
+ assert updated.status_code == 200
+ assert not await ImagesRepository(db_session).get_image_templates(image_id)
+ # Detaching a template does not silently remove the missing catalog row.
+ assert await ImagesRepository(db_session).get_image(image["path"]) is not None
diff --git a/tests/api/routes/controller/test_images.py b/tests/api/routes/controller/test_images.py
index 716897a30..ec498f763 100644
--- a/tests/api/routes/controller/test_images.py
+++ b/tests/api/routes/controller/test_images.py
@@ -110,15 +110,18 @@ def empty_image(tmpdir) -> str:
class TestImageRoutes:
async def test_create_image(self, app: FastAPI, client: AsyncClient, images_dir) -> None:
- Qemu.instance().create_disk_image = AsyncioMagicMock()
+ async def create_disk(path, options):
+ with open(path, "wb") as f:
+ f.write(b"QFI\xfb\x00\x00\x00")
+
+ Qemu.instance().create_disk_image = AsyncioMagicMock(side_effect=create_disk)
path = os.path.join(os.path.join(images_dir, "QEMU", "new_image.qcow2"))
- with open(path, "wb+") as f:
- f.write(b"QFI\xfb\x00\x00\x00")
image_name = os.path.basename(path)
response = await client.post(
app.url_path_for("create_qemu_image", image_path=image_name), json={"format": "qcow2", "size": 30}
)
assert response.status_code == status.HTTP_201_CREATED
+ assert os.path.isfile(path)
@pytest.mark.parametrize(
"image_type, fixture_name, valid_request",
diff --git a/tests/conftest.py b/tests/conftest.py
index 17acbb91a..bc9f7c823 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,5 @@
import pytest
+import asyncio
import pytest_asyncio
import tempfile
import shutil
@@ -101,12 +102,32 @@ async def base_client(app: FastAPI, db_session: AsyncSession) -> AsyncGenerator[
app.dependency_overrides[get_db_session] = _get_test_db
- async with AsyncClient(
- base_url="http://test-api",
- headers={"Content-Type": "application/json"},
- transport=ASGIWebSocketTransport(app=app),
- ) as async_client:
- yield async_client
+ # AnyIO cancel scopes must be entered and exited by the same task. Recent
+ # pytest-asyncio versions finalize yield fixtures in a different task.
+ ready = asyncio.get_running_loop().create_future()
+ finished = asyncio.Event()
+
+ async def own_client():
+ try:
+ async with AsyncClient(
+ base_url="http://test-api",
+ headers={"Content-Type": "application/json"},
+ transport=ASGIWebSocketTransport(app=app),
+ ) as async_client:
+ ready.set_result(async_client)
+ await finished.wait()
+ except BaseException as e:
+ if not ready.done():
+ ready.set_exception(e)
+ raise
+
+ owner = asyncio.create_task(own_client())
+ try:
+ yield await ready
+ finally:
+ finished.set()
+ await owner
+ app.dependency_overrides.pop(get_db_session, None)
@pytest_asyncio.fixture(loop_scope="class", scope="class")
diff --git a/tests/services/test_image_reconciliation.py b/tests/services/test_image_reconciliation.py
new file mode 100644
index 000000000..fa8e9beba
--- /dev/null
+++ b/tests/services/test_image_reconciliation.py
@@ -0,0 +1,706 @@
+import asyncio
+import hashlib
+import os
+import sys
+import threading
+import subprocess
+from types import SimpleNamespace
+from pathlib import Path
+from unittest.mock import patch, AsyncMock
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select, text, event
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.exc import SQLAlchemyError
+from watchdog.events import FileMovedEvent, FileDeletedEvent, FileModifiedEvent
+
+from gns3server.db.models import Base, Image, ImageSyncJob, Template
+from gns3server.db.models.images import image_template_map
+from gns3server.db.repositories.images import ImagesRepository
+from gns3server.services.image_reconciliation import ImageReconciliationService, InventoryEvents, enumerate_root
+from gns3server.utils.image_inventory import ImageLockBusy, image_lock, fingerprint
+from gns3server.utils.images import write_image, inspect_image_file, md5sum, InvalidImageError
+
+pytestmark = pytest.mark.asyncio
+QCOW = b"QFI\xfb\x00\x00\x00"
+
+
+@pytest_asyncio.fixture
+async def inventory(tmp_path, config):
+ engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'catalog.db'}")
+
+ @event.listens_for(engine.sync_engine, "connect")
+ def foreign_keys(connection, record):
+ connection.execute("PRAGMA foreign_keys=ON")
+
+ async with engine.connect() as conn:
+ with patch("gns3server.services.authentication.AuthService.hash_password", return_value="test"):
+ await conn.run_sync(Base.metadata.create_all)
+ await conn.commit()
+ service = ImageReconciliationService(engine, settle_seconds=0)
+ try:
+ yield service
+ finally:
+ await service.close()
+ await engine.dispose()
+
+
+def image_file(config, name="QEMU/image.qcow2", data=QCOW):
+ path = Path(config.settings.Server.images_path) / name
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(data)
+ return path
+
+
+async def scan(service, **kwargs):
+ job = await service.start(**kwargs)
+ await service.task
+ return await service.get_job(job["job_id"])
+
+
+async def rows(service):
+ async with AsyncSession(service.engine) as session:
+ return [image.asdict() for image in (await session.execute(select(Image))).scalars()]
+
+
+async def test_add_change_missing_restore_preserves_identity_and_template(inventory, config):
+ path = image_file(config)
+ assert (await scan(inventory))["counts"]["added"] == 1
+ original = (await rows(inventory))[0]
+ async with AsyncSession(inventory.engine) as session:
+ image = await session.get(Image, original["image_id"])
+ session.add(Template(name="Keep this reference", images=[image]))
+ await session.commit()
+ path.write_bytes(QCOW + b"changed")
+ assert (await scan(inventory))["counts"]["updated"] == 1
+ changed = (await rows(inventory))[0]
+ assert changed["image_id"] == original["image_id"]
+ assert changed["image_size"] == len(QCOW + b"changed")
+ assert changed["checksum"] == hashlib.md5(QCOW + b"changed", usedforsecurity=False).hexdigest()
+ path.unlink()
+ assert (await scan(inventory))["counts"]["missing"] == 1
+ assert (await rows(inventory))[0]["availability"] == "missing"
+ path.write_bytes(QCOW)
+ await scan(inventory)
+ restored = (await rows(inventory))[0]
+ assert restored["image_id"] == original["image_id"]
+ assert restored["availability"] == "available"
+ async with AsyncSession(inventory.engine) as session:
+ assert len((await session.execute(select(image_template_map))).all()) == 1
+
+
+async def test_fingerprints_skip_hashing_and_force_bypasses_sidecars(inventory, config):
+ path = image_file(config)
+ await scan(inventory)
+ with patch(
+ "gns3server.services.image_reconciliation.inspect_image_file", side_effect=AssertionError("unnecessary hash")
+ ):
+ result = await scan(inventory)
+ assert result["status"] == "completed"
+ assert result["counts"]["unchanged"] == 1
+ assert result["counts"]["bytes_hashed"] == 0
+ path.write_bytes(QCOW[:-1] + b"x") # same-size modification
+ sidecar = Path(str(path) + ".md5sum")
+ sidecar.write_text("0" * 32)
+ assert md5sum(str(path), cache_to_md5file=False) == "0" * 32
+ assert (
+ md5sum(str(path), cache_to_md5file=False, use_cache=False)
+ == hashlib.md5(path.read_bytes(), usedforsecurity=False).hexdigest()
+ )
+ result = await scan(inventory, force_checksum=True)
+ assert result["counts"]["bytes_hashed"] == len(QCOW)
+ assert (await rows(inventory))[0]["checksum"] == hashlib.md5(path.read_bytes(), usedforsecurity=False).hexdigest()
+ assert not sidecar.exists()
+
+
+async def test_dry_run_does_not_write_catalog_or_sidecar(inventory, config):
+ path = image_file(config)
+ sidecar = Path(str(path) + ".md5sum")
+ sidecar.write_text("0" * 32)
+ result = await scan(inventory, dry_run=True, force_checksum=True)
+ assert result["counts"]["added"] == 1
+ assert await rows(inventory) == []
+ assert sidecar.read_text() == "0" * 32
+ await scan(inventory)
+ before = await rows(inventory)
+ path.unlink()
+ assert (await scan(inventory, dry_run=True))["counts"]["missing"] == 1
+ assert await rows(inventory) == before
+
+
+@pytest.mark.parametrize("failure", ["missing_root", "partial_walk", "unreadable_file"])
+async def test_failed_scopes_do_not_remove_rows(inventory, config, failure):
+ path = image_file(config)
+ await scan(inventory)
+ root = str(config.settings.Server.images_path)
+ if failure == "missing_root":
+ os.rename(root, root + "-offline")
+ result = await scan(inventory)
+ elif failure == "partial_walk":
+ with patch(
+ "gns3server.services.image_reconciliation.enumerate_root",
+ return_value=({}, [{"path": root, "reason": "Permission denied"}]),
+ ):
+ result = await scan(inventory)
+ else:
+ with patch(
+ "gns3server.services.image_reconciliation.inspect_image_file", side_effect=PermissionError("denied")
+ ):
+ result = await scan(inventory, force_checksum=True)
+ assert result["status"] == "partial"
+ assert result["counts"]["missing"] == 0
+ assert len(await rows(inventory)) == 1
+ assert (await rows(inventory))[0]["availability"] == "unavailable"
+
+
+async def test_extra_nested_and_overlapping_roots(inventory, config, tmp_path):
+ extra = tmp_path / "extra"
+ extra.mkdir()
+ (extra / "image.qcow2").write_bytes(QCOW)
+ image_file(config, "QEMU/nested/image.qcow2")
+ image_file(config, "legacy.qcow2")
+ config.settings.Server.additional_images_paths = [str(extra), str(extra), config.settings.Server.images_path]
+ result = await scan(inventory)
+ assert result["counts"]["added"] == 3
+ assert len(await rows(inventory)) == 3
+
+
+async def test_rename_retains_old_reference(inventory, config):
+ path = image_file(config)
+ await scan(inventory)
+ old = (await rows(inventory))[0]
+ path.rename(path.with_name("renamed.qcow2"))
+ result = await scan(inventory)
+ assert result["counts"]["added"] == result["counts"]["missing"] == 1
+ assert next(row for row in await rows(inventory) if row["image_id"] == old["image_id"])["availability"] == "missing"
+
+
+async def test_symlink_images_are_imported_and_target_changes_detected(inventory, config):
+ target = image_file(config, ".storage/target.qcow2")
+ path = image_file(config)
+ path.unlink()
+ path.symlink_to(target)
+ assert fingerprint(path) == fingerprint(target)
+ result = await scan(inventory)
+ assert result["status"] == "completed", result
+ assert result["counts"]["added"] == 1
+ original = (await rows(inventory))[0]
+ assert original["path"] == str(path)
+ assert original["checksum"] == hashlib.md5(QCOW).hexdigest()
+ target.write_bytes(QCOW + b"changed")
+ result = await scan(inventory)
+ assert result["counts"]["updated"] == 1
+ updated = (await rows(inventory))[0]
+ assert updated["image_id"] == original["image_id"]
+ assert updated["checksum"] == hashlib.md5(target.read_bytes()).hexdigest()
+ assert path.is_symlink()
+
+
+@pytest.mark.parametrize("target_kind", ["directory", "missing"])
+async def test_fingerprint_rejects_symlinks_without_regular_file_targets(tmp_path, target_kind):
+ target = tmp_path / "target"
+ if target_kind == "directory":
+ target.mkdir()
+ link = tmp_path / "image.qcow2"
+ link.symlink_to(target, target_is_directory=target_kind == "directory")
+ with pytest.raises(OSError):
+ fingerprint(link)
+
+
+async def test_external_symlinks_hidden_files_and_libraries_are_not_imported(inventory, config, tmp_path):
+ outside = tmp_path / "outside.qcow2"
+ outside.write_bytes(QCOW)
+ path = image_file(config)
+ path.unlink()
+ path.symlink_to(outside)
+ image_file(config, ".hidden/hidden.qcow2")
+ image_file(config, "IOU/lib/library.so")
+ image_file(config, "QEMU/upload.tmp")
+ result = await scan(inventory)
+ assert result["counts"]["added"] == 0
+ assert outside.read_bytes() == QCOW
+
+
+@pytest.mark.parametrize(
+ "name,data,expected",
+ [
+ ("QEMU/disk.raw", b"raw image bytes", "qemu"),
+ ("IOS/ios.bin", b"\x7fELF\x01\x02\x01", "ios"),
+ ("IOU/iou.bin", b"\x7fELF\x02\x01\x01", "iou"),
+ ],
+)
+async def test_image_types_and_raw_policy(inventory, config, name, data, expected):
+ image_file(config, name, data)
+ result = await scan(inventory)
+ assert result["counts"]["added"] == 1
+ assert (await rows(inventory))[0]["image_type"] == expected
+
+
+async def test_invalid_replacement_not_usable(inventory, config):
+ path = image_file(config)
+ await scan(inventory)
+ path.write_bytes(b"bad")
+ assert (await scan(inventory))["counts"]["invalid"] == 1
+ async with AsyncSession(inventory.engine) as session:
+ repo = ImagesRepository(session)
+ assert await repo.get_image_by_checksum(hashlib.md5(QCOW, usedforsecurity=False).hexdigest()) is None
+ assert (await rows(inventory))[0]["checksum"] == hashlib.md5(QCOW, usedforsecurity=False).hexdigest()
+
+
+async def test_changed_during_inspection_is_deferred(inventory, config):
+ path = image_file(config)
+ original = inspect_image_file
+
+ def replace_after_hash(*args):
+ info = original(*args)
+ path.write_bytes(QCOW + b"still copying")
+ return info
+
+ with patch("gns3server.services.image_reconciliation.inspect_image_file", side_effect=replace_after_hash):
+ result = await scan(inventory)
+ assert result["counts"]["deferred"] == 1
+ assert await rows(inventory) == []
+ assert (await scan(inventory))["counts"]["added"] == 1
+
+
+async def test_per_file_database_failure_does_not_poison_next_file(inventory, config):
+ image_file(config, "QEMU/first.qcow2")
+ image_file(config, "QEMU/second.qcow2")
+ original = ImagesRepository.save_verified_image
+ calls = 0
+
+ async def fail_first(repo, info):
+ nonlocal calls
+ calls += 1
+ if calls == 1:
+ # Real failed SQL statement in the file's session.
+ await repo._db_session.execute(text("INSERT INTO nonexistent_image_table VALUES (1)"))
+ return await original(repo, info)
+
+ with patch.object(ImagesRepository, "save_verified_image", fail_first):
+ result = await scan(inventory)
+ assert result["status"] == "partial"
+ assert result["counts"]["added"] == 1
+ assert (await scan(inventory))["counts"]["added"] == 1
+ assert len(await rows(inventory)) == 2
+
+
+async def test_overlapping_jobs_rejected_and_interrupted_job_recovers(inventory, config):
+ path = image_file(config)
+ async with image_lock(str(path)):
+ job = await inventory.start()
+ other = ImageReconciliationService(inventory.engine, settle_seconds=0)
+ with pytest.raises(ImageLockBusy):
+ await other.start()
+ await inventory.task
+ async with AsyncSession(inventory.engine) as session:
+ stale = await session.get(ImageSyncJob, job["job_id"])
+ stale.status = "running"
+ await session.commit()
+ await scan(inventory)
+ assert (await inventory.get_job(job["job_id"]))["status"] == "interrupted"
+
+
+async def test_close_immediately_releases_lock_and_finishes_job(inventory, config):
+ image_file(config)
+ job = await inventory.start()
+ await inventory.close()
+ assert (await inventory.get_job(job["job_id"]))["status"] == "cancelled"
+ other = ImageReconciliationService(inventory.engine, settle_seconds=0)
+ try:
+ assert (await scan(other))["status"] == "completed"
+ finally:
+ await other.close()
+
+
+async def test_watcher_coalesces_move_delete_modify_events():
+ dirty = asyncio.Event()
+ handler = InventoryEvents(asyncio.get_running_loop(), dirty)
+ for fs_event in [
+ FileMovedEvent("/images/upload.tmp", "/images/new.qcow2"),
+ FileDeletedEvent("/images/deleted.qcow2"),
+ FileModifiedEvent("/images/changed.qcow2"),
+ ]:
+ dirty.clear()
+ handler.dispatch(fs_event)
+ await asyncio.sleep(0)
+ assert dirty.is_set()
+ dirty.clear()
+ handler.dispatch(FileModifiedEvent("/images/new.qcow2.md5sum"))
+ await asyncio.sleep(0)
+ assert not dirty.is_set()
+
+
+async def stream(data, chunk_size=1):
+ for offset in range(0, len(data), chunk_size):
+ yield data[offset : offset + chunk_size]
+
+
+async def test_upload_fragmented_header_and_missing_row_reuse(inventory, config):
+ path = image_file(config)
+ await scan(inventory)
+ old_id = (await rows(inventory))[0]["image_id"]
+ path.unlink()
+ await scan(inventory)
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ image = await write_image("QEMU/image.qcow2", str(path), stream(QCOW), ImagesRepository(session))
+ assert image.image_id == old_id
+ assert image.availability == "available"
+ assert path.read_bytes() == QCOW
+ assert not list(path.parent.glob("*.tmp"))
+
+
+async def test_upload_db_failure_is_recovered_by_scan(inventory, config):
+ path = Path(config.settings.Server.images_path) / "QEMU/orphan.qcow2"
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ repo = ImagesRepository(session)
+ with patch.object(repo, "save_verified_image", side_effect=SQLAlchemyError("commit failed")):
+ with pytest.raises(SQLAlchemyError):
+ await write_image("QEMU/orphan.qcow2", str(path), stream(QCOW), repo)
+ assert path.read_bytes() == QCOW
+ assert (await scan(inventory))["counts"]["added"] == 1
+
+
+async def test_concurrent_uploads_never_overwrite(inventory, config):
+ path = Path(config.settings.Server.images_path) / "QEMU/same.qcow2"
+
+ async def upload(data):
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ return await write_image("QEMU/same.qcow2", str(path), stream(data), ImagesRepository(session))
+
+ result = await asyncio.gather(upload(QCOW), upload(QCOW + b"different"), return_exceptions=True)
+ assert sum(isinstance(item, InvalidImageError) for item in result) == 1
+ assert len(await rows(inventory)) == 1
+ assert (await rows(inventory))[0]["checksum"] == hashlib.md5(path.read_bytes(), usedforsecurity=False).hexdigest()
+
+
+async def test_checksum_lookup_ignores_missing_duplicate(inventory, config):
+ first = image_file(config, "QEMU/first.qcow2")
+ second = image_file(config, "QEMU/second.qcow2")
+ await scan(inventory)
+ first.unlink()
+ # No sync is needed to reject a stale candidate at the point of use.
+ async with AsyncSession(inventory.engine) as session:
+ image = await ImagesRepository(session).get_image_by_checksum(
+ hashlib.md5(QCOW, usedforsecurity=False).hexdigest(), str(second.parent)
+ )
+ assert image.path == str(second)
+
+
+async def test_path_matching_does_not_interpret_sql_wildcards(inventory, config):
+ path = image_file(config, "QEMU/name_%/disk.qcow2")
+ image_file(config, "QEMU/name_AB/disk.qcow2")
+ await scan(inventory)
+ async with AsyncSession(inventory.engine) as session:
+ repo = ImagesRepository(session)
+ assert (await repo.get_image("name_%/disk.qcow2")).path == str(path)
+ assert await repo.get_image("me_%/disk.qcow2") is None
+
+
+async def test_migration_roundtrip_preserves_rows_and_relationships(inventory, config):
+ from alembic import command
+ from alembic.config import Config
+
+ image_file(config)
+ await scan(inventory)
+ original = (await rows(inventory))[0]
+ async with AsyncSession(inventory.engine) as session:
+ image = await session.get(Image, original["image_id"])
+ session.add(Template(name="Migration reference", images=[image]))
+ await session.commit()
+
+ def migrate(connection):
+ cfg = Config()
+ cfg.set_main_option("script_location", "gns3server:db_migrations")
+ cfg.attributes["connection"] = connection
+ command.stamp(cfg, "d9e8a2b7c401")
+ command.downgrade(cfg, "c7e4a9f1d2b6")
+ assert connection.execute(text("SELECT count(*) FROM image_template_map")).scalar_one() == 1
+ assert connection.execute(text("SELECT image_id FROM images")).scalar_one() == original["image_id"]
+ command.upgrade(cfg, "head")
+ assert connection.execute(text("PRAGMA foreign_key_check")).all() == []
+
+ async with inventory.engine.connect() as connection:
+ await connection.run_sync(migrate)
+ await connection.commit()
+ migrated = (await rows(inventory))[0]
+ assert migrated["availability"] == "unknown"
+ assert migrated["file_fingerprint"] is None
+ assert migrated["image_id"] == original["image_id"]
+ await scan(inventory)
+ assert (await rows(inventory))[0]["availability"] == "available"
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="Native Linux watcher integration")
+async def test_native_watcher_covers_extra_roots_and_stops(inventory, config, tmp_path):
+ extra = tmp_path / "extra-watch"
+ extra.mkdir()
+ config.settings.Server.additional_images_paths = [str(extra)]
+ config.settings.Server.auto_discover_images = True
+ await inventory._watch()
+ observer = inventory.observer
+ assert observer.is_alive()
+ inventory.dirty.clear()
+ temporary = extra / "incoming.tmp"
+ temporary.write_bytes(QCOW)
+ temporary.rename(extra / "finished.qcow2")
+ await asyncio.wait_for(inventory.dirty.wait(), 5)
+ assert (await scan(inventory))["counts"]["added"] == 1
+ inventory.dirty.clear()
+ (extra / "finished.qcow2").unlink()
+ await asyncio.wait_for(inventory.dirty.wait(), 5)
+ assert (await scan(inventory))["counts"]["missing"] == 1
+ await inventory.close()
+ assert not observer.is_alive()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX lock integration")
+async def test_image_lock_excludes_another_process(config, tmp_path):
+ path = str(tmp_path / "image.qcow2")
+ code = "import fcntl, sys; f = open(sys.argv[1], 'a+b'); fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)"
+ async with image_lock(path) as lock:
+ lock_path = lock._file.name
+ result = await asyncio.to_thread(subprocess.run, [sys.executable, "-c", code, lock_path], capture_output=True)
+ assert result.returncode != 0
+ assert b"BlockingIOError" in result.stderr
+ result = await asyncio.to_thread(subprocess.run, [sys.executable, "-c", code, lock_path], capture_output=True)
+ assert result.returncode == 0
+
+
+async def test_shutdown_during_hash_waits_for_worker_before_releasing_lock(inventory, config):
+ image_file(config)
+ entered = threading.Event()
+ exited = threading.Event()
+
+ def slow_hash(path, expected, raw, stopped):
+ entered.set()
+ stopped.wait(5)
+ exited.set()
+ raise InterruptedError("stopped")
+
+ with patch("gns3server.services.image_reconciliation.inspect_image_file", side_effect=slow_hash):
+ job = await inventory.start()
+ assert await asyncio.to_thread(entered.wait, 5)
+ await inventory.close()
+ assert exited.is_set()
+ assert (await inventory.get_job(job["job_id"]))["status"] == "cancelled"
+ assert await rows(inventory) == []
+
+
+async def test_appliance_fallback_repairs_existing_row_and_rejects_wrong_content(inventory, config):
+ from gns3server.controller.appliance_manager import ApplianceManager
+ from gns3server.controller.controller_error import ControllerError
+
+ path = image_file(config)
+ await scan(inventory)
+ old_id = (await rows(inventory))[0]["image_id"]
+ data = QCOW + b"replacement"
+ path.write_bytes(data)
+ appliance = SimpleNamespace(
+ images=[{"filename": path.name, "md5sum": hashlib.md5(data, usedforsecurity=False).hexdigest()}]
+ )
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ await ApplianceManager()._find_appliance_version_images(
+ appliance, {"images": {"hda_disk_image": path.name}}, ImagesRepository(session), str(path.parent)
+ )
+ assert (await rows(inventory))[0]["image_id"] == old_id
+ appliance.images[0]["md5sum"] = "0" * 32
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ with pytest.raises(ControllerError, match="checksum"):
+ await ApplianceManager()._find_appliance_version_images(
+ appliance, {"images": {"hda_disk_image": path.name}}, ImagesRepository(session), str(path.parent)
+ )
+
+
+async def test_legacy_path_spelling_preserves_identity(inventory, config):
+ path = image_file(config)
+ legacy_path = str(path.parent) + os.sep + "." + os.sep + path.name
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ original = await ImagesRepository(session).add_image(path.name, "qemu", len(QCOW), legacy_path, "0" * 32, "md5")
+ original_id = original.image_id
+ await scan(inventory)
+ images = await rows(inventory)
+ assert len(images) == 1
+ assert images[0]["image_id"] == original_id
+ assert images[0]["path"] == legacy_path
+ assert images[0]["checksum"] == hashlib.md5(QCOW, usedforsecurity=False).hexdigest()
+
+
+async def test_root_disappearing_during_scan_does_not_mark_images_missing(inventory, config):
+ first = image_file(config, "QEMU/first.qcow2")
+ image_file(config, "QEMU/second.qcow2")
+ await scan(inventory)
+ first.unlink()
+ root = config.settings.Server.images_path
+ original = inventory._file
+
+ async def disconnect_root(*args):
+ await original(*args)
+ os.rename(root, root + "-disconnected")
+
+ with patch.object(inventory, "_file", side_effect=disconnect_root):
+ result = await scan(inventory)
+ assert result["status"] == "partial"
+ assert result["counts"]["missing"] == 0
+ images = await rows(inventory)
+ assert next(image for image in images if image["path"] == str(first))["availability"] == "unavailable"
+
+
+async def test_polling_recovers_crashed_job_without_starting_a_scan(inventory, config):
+ image_file(config)
+ job = await scan(inventory)
+ async with AsyncSession(inventory.engine) as session:
+ row = await session.get(ImageSyncJob, job["job_id"])
+ row.status = "running"
+ await session.commit()
+ assert not config.settings.Server.auto_discover_images
+ assert (await inventory.get_job(job["job_id"]))["status"] == "interrupted"
+
+
+async def test_periodic_scan_recovers_without_watcher_events(inventory, config):
+ config.settings.Server.auto_discover_images = True
+ image_file(config)
+ # Use a short interval and initial wait in the service, leaving asyncio's
+ # event-loop scheduling and database operations intact.
+ real_sleep = asyncio.sleep
+
+ async def short_sleep(seconds):
+ await real_sleep(0.01)
+
+ scans = []
+ original_start = inventory.start
+
+ async def observed_start(*args, **kwargs):
+ job = await original_start(*args, **kwargs)
+ scans.append(job["job_id"])
+ return job
+
+ async def missed_event(*args, **kwargs):
+ # Consume/close the unused Event.wait coroutine, then simulate a timeout.
+ args[0].close()
+ await real_sleep(0.01)
+ raise asyncio.TimeoutError
+
+ with (
+ patch.object(inventory, "_watch", new=AsyncMock()),
+ patch.object(inventory, "start", side_effect=observed_start),
+ patch("gns3server.services.image_reconciliation.asyncio.sleep", side_effect=short_sleep),
+ patch("gns3server.services.image_reconciliation.asyncio.wait_for", side_effect=missed_event),
+ ):
+ inventory.start_background()
+ for _ in range(500):
+ if len(scans) >= 2:
+ break
+ await real_sleep(0.01)
+ await inventory.close()
+ assert len(scans) >= 2
+ assert len(await rows(inventory)) == 1
+
+
+async def test_unchanged_collection_avoids_hashing_and_keeps_loop_responsive(inventory, config):
+ import time
+
+ for index in range(200):
+ image_file(config, f"QEMU/bulk/{index}.qcow2", QCOW + b"x" * 65536)
+ samples = []
+ done = asyncio.Event()
+
+ async def heartbeat():
+ previous = time.monotonic()
+ while not done.is_set():
+ await asyncio.sleep(0.01)
+ current = time.monotonic()
+ samples.append(current - previous)
+ previous = current
+
+ pulse = asyncio.create_task(heartbeat())
+ try:
+ first = await scan(inventory)
+ second = await scan(inventory)
+ finally:
+ done.set()
+ await pulse
+ assert first["counts"]["added"] == 200
+ assert second["counts"]["unchanged"] == 200
+ assert second["counts"]["bytes_hashed"] == 0
+ assert len(samples) > 2
+ # Generous stall threshold; this checks responsiveness rather than host speed.
+ assert max(samples) < 2
+
+
+async def test_startup_upgrades_unversioned_existing_catalog(inventory, config, monkeypatch):
+ from alembic import command
+ from alembic.config import Config
+ from fastapi import FastAPI
+ from gns3server.db.tasks import connect_to_db, disconnect_from_db
+
+ image_file(config)
+ await scan(inventory)
+ original_id = (await rows(inventory))[0]["image_id"]
+
+ def remove_inventory_schema(connection):
+ cfg = Config()
+ cfg.set_main_option("script_location", "gns3server:db_migrations")
+ cfg.attributes["connection"] = connection
+ command.stamp(cfg, "d9e8a2b7c401")
+ command.downgrade(cfg, "c7e4a9f1d2b6")
+ connection.execute(text("DROP TABLE alembic_version"))
+
+ async with inventory.engine.connect() as connection:
+ await connection.run_sync(remove_inventory_schema)
+ await connection.commit()
+ monkeypatch.setenv("GNS3_DATABASE_URI", str(inventory.engine.url))
+ app = FastAPI()
+ await connect_to_db(app)
+ try:
+ restored = (await rows(inventory))[0]
+ assert restored["image_id"] == original_id
+ assert restored["availability"] == "unknown"
+ finally:
+ await disconnect_from_db(app)
+
+
+@pytest.mark.parametrize("operation", ["delete", "prune"])
+@pytest.mark.parametrize("reuse_id", [False, True])
+async def test_delete_preserves_replacement_created_while_waiting_for_lock(inventory, config, operation, reuse_id):
+ from contextlib import asynccontextmanager
+ from sqlalchemy import delete
+ from gns3server.api.routes.controller.images import delete_image
+ from gns3server.controller.controller_error import ControllerError
+
+ path = image_file(config)
+ await scan(inventory)
+ original = (await rows(inventory))[0]
+ replacement = QCOW + b"replacement uploaded by another request"
+ replacement_id = original["image_id"] if reuse_id else original["image_id"] + 100
+
+ @asynccontextmanager
+ async def concurrent_replacement(locked_path):
+ assert locked_path == str(path)
+ async with AsyncSession(inventory.engine) as writer:
+ await writer.execute(delete(Image).where(Image.image_id == original["image_id"]))
+ path.write_bytes(replacement)
+ info = inspect_image_file(str(path))
+ info["filename"] = info.pop("image_name")
+ writer.add(Image(image_id=replacement_id, **info))
+ await writer.commit()
+ yield
+
+ module = "gns3server.api.routes.controller.images" if operation == "delete" else "gns3server.db.repositories.images"
+ with (
+ patch(module + ".image_lock", concurrent_replacement),
+ patch(
+ "gns3server.api.routes.controller.images.Controller.instance",
+ return_value=SimpleNamespace(find_projects_using_image=lambda filename: []),
+ ),
+ ):
+ async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+ repository = ImagesRepository(session)
+ if operation == "delete":
+ with pytest.raises(ControllerError, match="changed"):
+ await delete_image(str(path), repository)
+ else:
+ assert await repository.prune_images() == 0
+ assert path.read_bytes() == replacement
+ assert (await rows(inventory))[0]["image_id"] == replacement_id