From 046dff4d0819d2224ccb6a0638947a9f4dd7ba33 Mon Sep 17 00:00:00 2001 From: Cristi Date: Mon, 28 Sep 2026 16:01:36 +0300 Subject: [PATCH 1/5] fix(images): reconcile the image inventory with filesystem changes --- docs/README.md | 10 + gns3server/api/routes/controller/images.py | 127 ++-- gns3server/controller/appliance_manager.py | 12 +- gns3server/core/tasks.py | 32 +- gns3server/db/models/__init__.py | 2 +- gns3server/db/models/images.py | 19 +- gns3server/db/repositories/images.py | 160 ++++- gns3server/db/repositories/templates.py | 40 +- gns3server/db/tasks.py | 169 +---- ...a2b7c401_image_inventory_reconciliation.py | 50 ++ gns3server/schemas/__init__.py | 2 +- gns3server/schemas/config.py | 2 + gns3server/schemas/controller/images.py | 23 + gns3server/schemas/controller/settings.py | 1 + gns3server/services/image_reconciliation.py | 450 +++++++++++++ gns3server/services/templates.py | 3 +- gns3server/utils/image_inventory.py | 96 +++ gns3server/utils/images.py | 144 ++-- .../api/routes/controller/test_image_sync.py | 130 ++++ tests/api/routes/controller/test_images.py | 11 +- tests/conftest.py | 33 +- tests/services/test_image_reconciliation.py | 628 ++++++++++++++++++ 22 files changed, 1809 insertions(+), 335 deletions(-) create mode 100644 gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py create mode 100644 gns3server/services/image_reconciliation.py create mode 100644 gns3server/utils/image_inventory.py create mode 100644 tests/api/routes/controller/test_image_sync.py create mode 100644 tests/services/test_image_reconciliation.py 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 fe27c20d8..d639e69fa 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 @@ -36,6 +38,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, @@ -51,6 +55,37 @@ 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, @@ -76,33 +111,32 @@ 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( "", @@ -111,7 +145,8 @@ async def create_qemu_image( ) async def get_images( images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)), - image_type: Optional[schemas.ImageType] = None + image_type: Optional[schemas.ImageType] = None, + availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None ) -> List[schemas.Image]: """ Return all images. @@ -119,7 +154,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( @@ -149,10 +184,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") @@ -168,7 +200,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 +241,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( @@ -232,6 +269,9 @@ 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 @@ -318,11 +358,20 @@ 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 e411b9852..7f6969d52 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 @@ -184,13 +185,14 @@ 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 0b745a067..dfea5e7ab 100644 --- a/gns3server/core/tasks.py +++ b/gns3server/core/tasks.py @@ -26,22 +26,22 @@ 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,19 +66,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 +95,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 529afe4bf..73b872087 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 relationship from .base import Base, BaseTable, GUID @@ -40,4 +40,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 8f6c64dc9..a34c116f8 100644 --- a/gns3server/db/repositories/images.py +++ b/gns3server/db/repositories/images.py @@ -16,10 +16,14 @@ # along with this program. If not, see . import os +import asyncio +from datetime import datetime, timezone -from typing import Optional, List +from typing import Optional, List, Callable from sqlalchemy import select, delete, update 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,18 +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)) + where(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: str = None) -> Optional[models.Image]: @@ -55,17 +62,28 @@ 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. """ @@ -74,6 +92,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 result.scalars().all() @@ -89,7 +109,8 @@ class ImagesRepository(BaseRepository): result = await self._db_session.execute(query) return 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. """ @@ -101,11 +122,17 @@ class ImagesRepository(BaseRepository): image_size=image_size, path=path, checksum=checksum, - checksum_algorithm=checksum_algorithm + 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 @@ -125,15 +152,69 @@ 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) -> 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 + + 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 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: query = delete(models.Image).where(models.Image.filename == image_name) @@ -141,7 +222,8 @@ class ImagesRepository(BaseRepository): await self._db_session.commit() return result.rowcount > 0 - async def prune_images(self, skip_images: list[str] = None) -> int: + async def prune_images(self, skip_images: list[str] = None, + is_in_use: Optional[Callable] = None) -> int: """ Prune images not attached to any template. """ @@ -149,18 +231,32 @@ class ImagesRepository(BaseRepository): 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 a08b58053..cb2d90372 100644 --- a/gns3server/db/repositories/templates.py +++ b/gns3server/db/repositories/templates.py @@ -29,6 +29,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__) @@ -129,17 +131,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)) + where(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() @@ -160,18 +167,23 @@ 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, diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index 285d95b54..4072644df 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 @@ -138,7 +133,14 @@ 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") @@ -182,144 +184,24 @@ async def get_computes(app: FastAPI) -> List[dict]: 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 in discovered_images: - log.info(f"Adding discovered image '{image['path']}' to the database") - try: - await images_repository.add_image(**image) - except SQLAlchemyError as e: - log.warning(f"Error while adding image '{image['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 - 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.BaseEventLoop, **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.BaseEventLoop, - 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() - 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}") + from gns3server.utils.image_inventory import image_lock + for path in updated_disks: + async with image_lock(path): + async with AsyncSession(app.state._db_engine, expire_on_commit=False) as session: + repository = ImagesRepository(session) + image = await repository.get_image(path) + if image: + info = await read_image_info(path, 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]: @@ -415,4 +297,3 @@ async def get_user_llm_config_full(user_id: str, app: FastAPI) -> Optional[dict] except Exception as e: log.error(f"Failed to retrieve LLM config for user {user_id}: {e}", exc_info=True) return None - 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..aaef4e91e --- /dev/null +++ b/gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py @@ -0,0 +1,50 @@ +"""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 bc8cca5b9..af96a5f7a 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -23,7 +23,7 @@ from .version import Version from .controller.links import LinkCreate, LinkUpdate, Link, UDPPortInfo, EthernetPortInfo, LinkCapture, MarkerCreate, MarkerUpdate, MarkerDefinitionCreate from .controller.computes import ComputeCreate, ComputeUpdate, ComputeVirtualBoxVM, ComputeVMwareVM, ComputeDockerImage, AutoIdlePC, 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 227129b28..c92a6516d 100644 --- a/gns3server/schemas/config.py +++ b/gns3server/schemas/config.py @@ -197,6 +197,8 @@ class ServerSettings(BaseModel): 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( diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py index 417bc086f..328fd2c4a 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 @@ -38,7 +40,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 73bace5bf..fd84cf398 100644 --- a/gns3server/schemas/controller/settings.py +++ b/gns3server/schemas/controller/settings.py @@ -117,6 +117,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..cc7db82cf --- /dev/null +++ b/gns3server/services/image_reconciliation.py @@ -0,0 +1,450 @@ +"""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): + 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): + 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 66b793566..2d2771d26 100644 --- a/gns3server/services/templates.py +++ b/gns3server/services/templates.py @@ -276,7 +276,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..fcab6fe09 --- /dev/null +++ b/gns3server/utils/image_inventory.py @@ -0,0 +1,96 @@ +"""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=False) + 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 a1074191d..1c21b0886 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 @@ -123,31 +126,40 @@ def get_builtin_disks() -> List[str]: return builtin_disks -async def read_image_info(path: str, expected_image_type: str = 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) | getattr(os, "O_NOFOLLOW", 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, 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) -> List[dict]: @@ -239,7 +251,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 @@ -258,7 +270,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() @@ -315,6 +327,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': @@ -343,57 +359,53 @@ 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, 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, " - f"please choose a different name or remove the existing image") - - checksum = checksum.hexdigest() - image_dir = os.path.dirname(image_path) - duplicate_image = await images_repo.get_image_by_checksum(checksum, image_dir) - if duplicate_image: - raise InvalidImageError(f"Image '{duplicate_image.filename}' with the " - f"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) + async with image_lock(image_path): + if os.path.lexists(image_path): + raise InvalidImageError(f"File '{image_path}' already exists, " + f"please choose a different name or remove the existing image") + checksum = checksum.hexdigest() + duplicate_image = await images_repo.get_image_by_checksum(checksum, 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. + return await images_repo.save_verified_image(dict( + image_name=image_name, image_type=image_type, image_size=image_size, + path=image_path, checksum=checksum, checksum_algorithm="md5", + file_fingerprint=fingerprint(image_path))) 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, - checksum_algorithm="md5" - ) 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..13d9fa7ca --- /dev/null +++ b/tests/api/routes/controller/test_image_sync.py @@ -0,0 +1,130 @@ +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 ea0b2f587..fc50b5320 100644 --- a/tests/api/routes/controller/test_images.py +++ b/tests/api/routes/controller/test_images.py @@ -111,14 +111,17 @@ 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", @@ -473,4 +476,4 @@ class TestImageRoutes: templates = await templates_repo.get_templates() assert len(templates) == 1 assert templates[0].name == "Empty VM" - await templates_repo.delete_template(templates[0].template_id) \ No newline at end of file + await templates_repo.delete_template(templates[0].template_id) diff --git a/tests/conftest.py b/tests/conftest.py index b0994ebb8..e9865007d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import pytest +import asyncio import pytest_asyncio import tempfile import shutil @@ -100,12 +101,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..94d4e7946 --- /dev/null +++ b/tests/services/test_image_reconciliation.py @@ -0,0 +1,628 @@ +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").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()).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()).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_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).hexdigest()) is None + assert (await rows(inventory))[0]["checksum"] == hashlib.md5(QCOW).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 event in [FileMovedEvent("/images/upload.tmp", "/images/new.qcow2"), + FileDeletedEvent("/images/deleted.qcow2"), FileModifiedEvent("/images/changed.qcow2")]: + dirty.clear() + handler.dispatch(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()).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).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).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).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 From 2e127759113ed16fbbac455a93857fd214d43529 Mon Sep 17 00:00:00 2001 From: Cristi Date: Tue, 29 Sep 2026 15:42:49 +0300 Subject: [PATCH 2/5] fix(images): Fix symlink support in image inventory --- gns3server/utils/image_inventory.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gns3server/utils/image_inventory.py b/gns3server/utils/image_inventory.py index e222d5a69..78943ba11 100644 --- a/gns3server/utils/image_inventory.py +++ b/gns3server/utils/image_inventory.py @@ -20,7 +20,7 @@ def contained_path(path, root): def fingerprint(path): - info = os.stat(path, follow_symlinks=False) + 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) From 87df09c00ca4b3425fa32128599a636a3fe5cfcc Mon Sep 17 00:00:00 2001 From: Cristi Date: Tue, 29 Sep 2026 20:16:22 +0300 Subject: [PATCH 3/5] fix(images): Fix symlinked image imports and add regression tests --- gns3server/utils/images.py | 2 +- tests/services/test_image_reconciliation.py | 34 ++++++++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py index a1b208c5c..58acc8b1e 100644 --- a/gns3server/utils/images.py +++ b/gns3server/utils/images.py @@ -132,7 +132,7 @@ def get_builtin_disks() -> List[str]: 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) | getattr(os, "O_NOFOLLOW", 0) + 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: diff --git a/tests/services/test_image_reconciliation.py b/tests/services/test_image_reconciliation.py index cf0c41d6a..fa8e9beba 100644 --- a/tests/services/test_image_reconciliation.py +++ b/tests/services/test_image_reconciliation.py @@ -176,7 +176,39 @@ async def test_rename_retains_old_reference(inventory, config): assert next(row for row in await rows(inventory) if row["image_id"] == old["image_id"])["availability"] == "missing" -async def test_symlinks_hidden_files_and_libraries_are_not_imported(inventory, config, tmp_path): +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) From 285af128f2c8a2f6bce485c1a4648432cfedb92d Mon Sep 17 00:00:00 2001 From: grossmj Date: Thu, 1 Oct 2026 14:53:54 +0200 Subject: [PATCH 4/5] fix: ignore S324 error in tests --- pyproject.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index f979b5eb3..9c76a1d74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -277,6 +277,8 @@ extend-immutable-calls = [ "tests/test_utils.py" = ["F403", "F405"] "tests/utils/test_asyncio.py" = ["B017"] "tests/utils/test_packet_filter_validation.py" = ["W292"] +"tests/tests/services/test_image_reconciliation.py" = ["S324"] + [tool.mypy] From 1456d36bf06fce0eaab412db715f6ee3bd6497e8 Mon Sep 17 00:00:00 2001 From: grossmj Date: Thu, 1 Oct 2026 15:22:17 +0200 Subject: [PATCH 5/5] chore(tests): fix mypy errors --- gns3server/api/routes/controller/images.py | 2 +- gns3server/db/repositories/images.py | 7 ++++--- gns3server/db/tasks.py | 2 +- gns3server/utils/images.py | 15 +++++++++------ pyproject.toml | 4 +--- 5 files changed, 16 insertions(+), 14 deletions(-) diff --git a/gns3server/api/routes/controller/images.py b/gns3server/api/routes/controller/images.py index 6141e4152..0496e29ee 100644 --- a/gns3server/api/routes/controller/images.py +++ b/gns3server/api/routes/controller/images.py @@ -110,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. diff --git a/gns3server/db/repositories/images.py b/gns3server/db/repositories/images.py index e2d0f2401..1d52295e7 100644 --- a/gns3server/db/repositories/images.py +++ b/gns3server/db/repositories/images.py @@ -154,7 +154,7 @@ 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) -> models.Image: + 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 @@ -217,11 +217,12 @@ class ImagesRepository(BaseRepository): 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 result.rowcount > 0 + return cast(CursorResult, result).rowcount > 0 async def delete_image(self, image_path: str) -> bool: """ @@ -246,7 +247,7 @@ 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, is_in_use: Optional[Callable] = 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. """ diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index b3f762021..64f3f2673 100644 --- a/gns3server/db/tasks.py +++ b/gns3server/db/tasks.py @@ -203,7 +203,7 @@ async def update_disk_checksums(updated_disks: List[str]) -> None: repository = ImagesRepository(db_session) image = await repository.get_image(path) if image: - info = await read_image_info(path, image.image_type, allow_raw_image=True) + info = await read_image_info(path, str(image.image_type), allow_raw_image=True) try: os.unlink(path + ".md5sum") except FileNotFoundError: diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py index fa4be5cb7..e2c458e1b 100644 --- a/gns3server/utils/images.py +++ b/gns3server/utils/images.py @@ -164,7 +164,7 @@ def inspect_image_file(path, expected_image_type=None, allow_raw_image=False, st ) -async def read_image_info(path: str, expected_image_type: str = None, allow_raw_image=False) -> dict: +async def read_image_info(path: str, expected_image_type: str | None = None, allow_raw_image=False) -> dict: try: return await asyncio.to_thread(inspect_image_file, path, expected_image_type, allow_raw_image) except OSError as e: @@ -376,7 +376,7 @@ async def write_image( 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, prefix, allow_raw_image or not check_image_header) + 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)) @@ -399,8 +399,8 @@ async def write_image( raise InvalidImageError( f"File '{image_path}' already exists, please choose a different name or remove the existing image" ) - checksum = checksum.hexdigest() - duplicate_image = await images_repo.get_image_by_checksum(checksum, os.path.dirname(image_path)) + 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 " @@ -410,17 +410,20 @@ async def write_image( 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. - return await images_repo.save_verified_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, + checksum=checksum_str, checksum_algorithm="md5", file_fingerprint=fingerprint(image_path), ) ) + if image is None: + raise InvalidImageError(f"Failed to save image '{image_name}' to database") + return image finally: try: if os.path.exists(tmp_path): diff --git a/pyproject.toml b/pyproject.toml index 9c76a1d74..07c46dc7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -277,9 +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/tests/services/test_image_reconciliation.py" = ["S324"] - - +"tests/services/test_image_reconciliation.py" = ["S324"] [tool.mypy] files = ["gns3server"]