Merge branch '3.1' into fix/duplicate-device-configs-merge

tests/api/routes/controller/test_image_subfolders.py

@@ -0,0 +1,254 @@
+"""Uploads into server-selected type folders and restricted relative subfolders."""
+
+import hashlib
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.exc import SQLAlchemyError
+
+from gns3server.controller.controller_error import ControllerError
+from gns3server.db.repositories.images import ImagesRepository
+from tests.api.routes.controller.test_image_sync import sync_service  # noqa: F401 - pytest fixture discovery
+
+pytestmark = pytest.mark.asyncio
+QCOW = b"QFI\xfb\x00\x00\x00"
+QCOW_CHECKSUM = hashlib.md5(QCOW, usedforsecurity=False).hexdigest()
+
+
+async def upload(client, filename="router.qcow2", **kwargs):
+    return await client.post(f"/v3/images/upload/{filename}", content=kwargs.pop("content", QCOW), **kwargs)
+
+
+class TestImageSubfolders:
+    @pytest.mark.parametrize(
+        "content,folder,subfolder,filename",
+        [
+            (QCOW, "QEMU", "Vendor/Version 1", "router.bin"),
+            (b"\x7fELF\x01\x02\x01", "IOS", "Vendor/Version 1", "router.bin"),
+            (b"\x7fELF\x02\x01\x01", "IOU", "Vendor/Version 1", "router.bin"),
+            (QCOW, "QEMU", "TACACS", "tacacs.qcow2"),
+        ],
+    )
+    async def test_server_detects_type_and_sync_restores_nested_image(
+        self,
+        client,
+        images_dir,
+        db_session,
+        sync_service,  # noqa: F811 - pytest fixture imported for discovery
+        content,
+        folder,
+        subfolder,
+        filename,
+    ):
+        response = await upload(
+            client, filename, params={"subdirectory": subfolder, "install_appliances": False}, content=content
+        )
+        assert response.status_code == 201, response.text
+        path = Path(images_dir) / folder / subfolder / filename
+        assert response.json()["path"] == str(path)
+        assert path.read_bytes() == content
+        assert not (Path(images_dir) / folder / filename).exists()
+        repository = ImagesRepository(db_session)
+        image_id = (await repository.get_image(str(path))).image_id
+        path.unlink()
+        await client.post("/v3/images/sync", json={})
+        await sync_service.task
+        assert (await repository.get_image(str(path), refresh=True)).availability == "missing"
+        path.write_bytes(content)  # Same filesystem operation used by an SFTP copy.
+        sibling = path.with_name("copied-over-ssh.bin")
+        sibling.write_bytes(content + b"new")
+        await client.post("/v3/images/sync", json={})
+        await sync_service.task
+        restored = await repository.get_image(str(path), refresh=True)
+        assert restored.image_id == image_id
+        assert restored.availability == "available"
+        assert await repository.get_image(str(sibling)) is not None
+
+    @pytest.mark.parametrize(
+        "folder",
+        [
+            "../escape",
+            "/tmp/escape",
+            "C:/escape",
+            "C:\\escape",
+            "a/../b",
+            "a//b",
+            "a/",
+            ".hidden",
+            "a/%2e%2e",
+            "a/%252e%252e",
+            "a/\x00bad",
+            "a/CON",
+            "a/NUL.txt",
+            "lib",
+            "lib64",
+            "a/b.tmp",
+            "a/b.md5sum",
+            "a/b.",
+            "a/b ",
+            "/".join(["a"] * 9),
+            "a" * 65,
+        ],
+    )
+    async def test_rejects_unsafe_subfolders(self, client, images_dir, folder):
+        before = set(Path(images_dir).rglob("*"))
+        response = await upload(client, params={"subdirectory": folder})
+        assert response.status_code == 400, response.text
+        assert set(Path(images_dir).rglob("*")) == before
+
+    @pytest.mark.parametrize("filename", ["QEMU/router.qcow2", ".hidden", "NUL.qcow2", "router:stream.qcow2"])
+    async def test_subfolder_upload_requires_plain_filename(self, client, filename):
+        response = await upload(client, filename, params={"subdirectory": "Vendor"})
+        assert response.status_code == 400
+
+    @pytest.mark.parametrize("link,inside", [("QEMU/link", False), ("QEMU/link", True), ("QEMU", False)])
+    async def test_refuses_symlinks_without_creating_children(self, client, images_dir, tmp_path, link, inside):
+        target = Path(images_dir) / "safe-target" if inside else tmp_path / "outside"
+        target.mkdir()
+        link_path = Path(images_dir) / link
+        if link_path.is_dir():
+            link_path.rmdir()
+        link_path.symlink_to(target, target_is_directory=True)
+        folder = "link/child" if link.endswith("/link") else "Vendor/Version"
+        response = await upload(client, params={"subdirectory": folder})
+        assert response.status_code == 409
+        assert not list(target.iterdir())
+
+    async def test_duplicate_checks_are_scoped_to_destination(self, client, images_dir):
+        for folder in ("VendorA", "VendorB"):
+            response = await upload(
+                client, "same.qcow2", params={"subdirectory": folder}, headers={"X-MD5-Checksum": QCOW_CHECKSUM}
+            )
+            assert response.status_code == 201
+        response = await upload(client, "duplicate.qcow2", params={"subdirectory": "VendorA"})
+        assert response.status_code == 409
+        assert not (Path(images_dir) / "QEMU/VendorA/duplicate.qcow2").exists()
+        response = await upload(client, "same.qcow2", params={"subdirectory": "VendorA"}, content=QCOW + b"changed")
+        assert response.status_code == 409
+        assert (Path(images_dir) / "QEMU/VendorA/same.qcow2").read_bytes() == QCOW
+
+    async def test_empty_subfolder_keeps_default_location(self, client, images_dir):
+        response = await upload(client, "default.qcow2", params={"subdirectory": ""})
+        assert response.status_code == 201
+        assert response.json()["path"] == str(Path(images_dir) / "QEMU/default.qcow2")
+
+    async def test_subfolder_creation_requires_authenticated_allocate_permission(self, client, images_dir, test_user):
+        from gns3server.services import auth_service
+        from gns3server.services.authentication import DEFAULT_JWT_SECRET_KEY
+
+        response = await upload(client, params={"subdirectory": "Unauthorized"}, headers={"Authorization": ""})
+        assert response.status_code == 401
+        token = auth_service.create_access_token(test_user.username, secret_key=DEFAULT_JWT_SECRET_KEY)
+        with patch(
+            "gns3server.db.repositories.rbac.RbacRepository.check_user_has_privilege", new=AsyncMock(return_value=False)
+        ):
+            response = await upload(
+                client, params={"subdirectory": "Unauthorized"}, headers={"Authorization": f"Bearer {token}"}
+            )
+        assert response.status_code == 403
+        assert not (Path(images_dir) / "QEMU/Unauthorized").exists()
+
+
+@pytest.mark.parametrize(
+    "results", [[], [{"status": "created", "name": "Router"}], [{"status": "skipped", "reason": "Missing disk"}]]
+)
+async def test_upload_reports_template_outcomes(client, controller, images_dir, results):
+    with patch.object(
+        controller.appliance_manager, "install_appliances_from_image", new=AsyncMock(return_value=results)
+    ) as install:
+        response = await upload(client, "custom.qcow2", params={"install_appliances": True, "subdirectory": "TACACS"})
+    assert response.status_code == 201, response.text
+    assert Path(response.json()["path"]).read_bytes() == QCOW
+    install.assert_awaited_once()
+    assert install.call_args.args[1] == QCOW_CHECKSUM
+    if results:
+        assert response.json()["template_results"][0]["status"] == results[0]["status"]
+    else:
+        assert (
+            "No compatible appliance definition found in the server's catalog for this image."
+            in response.json()["template_results"][0]["reason"]
+        )
+
+
+@pytest.mark.parametrize(
+    "rollback,error",
+    [(False, ControllerError("Missing disk")), (True, SQLAlchemyError("Template transaction failed"))],
+    ids=["missing-dependency", "database-rollback"],
+)
+async def test_template_failure_preserves_uploaded_image(client, controller, images_dir, db_session, rollback, error):
+    async def fail(*args):
+        if rollback:
+            await db_session.rollback()
+        raise error
+
+    with patch.object(controller.appliance_manager, "install_appliances_from_image", side_effect=fail):
+        response = await upload(client, "custom.qcow2", params={"install_appliances": True})
+    assert response.status_code == 201, response.text
+    assert Path(response.json()["path"]).read_bytes() == QCOW
+    assert str(error) in response.json()["template_results"][0]["reason"]
+
+
+async def test_normal_upload_omits_template_results(client, images_dir):
+    response = await upload(client, "custom.qcow2")
+    assert response.status_code == 201, response.text
+    assert "template_results" not in response.json()
+
+
+async def test_pre_upload_compatibility_is_read_only(client, controller, images_dir):
+    with patch.object(
+        controller.appliance_manager,
+        "check_image_compatibility",
+        new=AsyncMock(return_value=[{"checksum": "a" * 32, "matches": []}]),
+    ) as check:
+        response = await client.post("/v3/images/compatibility", json={"checksums": ["a" * 32]})
+    assert response.status_code == 200, response.text
+    assert response.json() == [{"checksum": "a" * 32, "matches": []}]
+    check.assert_awaited_once()
+    assert list(Path(images_dir).rglob("*.qcow2")) == []
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"checksums": []},
+        {"checksums": ["../escape"]},
+        {"checksums": ["a" * 33]},
+        {"checksums": ["a" * 32] * 1001},
+        {"checksums": ["a" * 32], "path": "/tmp"},
+    ],
+)
+async def test_compatibility_rejects_invalid_input(client, payload):
+    response = await client.post("/v3/images/compatibility", json=payload)
+    assert response.status_code == 422
+
+
+async def test_catalog_size_filter_is_available_before_upload(client, controller):
+    with patch.object(
+        controller.appliance_manager,
+        "image_compatibility_catalog",
+        return_value={"image_sizes": [123], "has_unknown_sizes": False},
+    ):
+        response = await client.get("/v3/images/compatibility/catalog")
+    assert response.status_code == 200, response.text
+    assert response.json() == {"image_sizes": [123], "has_unknown_sizes": False}
+
+
+async def test_appliance_dependency_download_preserves_nested_destination(controller, images_dir, db_session):
+    from types import SimpleNamespace
+
+    async def content():
+        yield QCOW
+
+    response = SimpleNamespace(status=200, content=SimpleNamespace(iter_any=content))
+    context = AsyncMock()
+    context.__aenter__.return_value = response
+    destination = Path(images_dir) / "QEMU/Vendor/Version"
+    with patch("gns3server.controller.appliance_manager.HTTPClient.get", return_value=context):
+        image = await controller.appliance_manager._download_image(
+            str(destination), "dependency.qcow2", "qemu", "https://example.com/disk", ImagesRepository(db_session)
+        )
+    expected = destination / "dependency.qcow2"
+    assert image.path == str(expected)
+    assert expected.read_bytes() == QCOW
+    assert not (Path(images_dir) / "QEMU/dependency.qcow2").exists()

tests/api/routes/controller/test_templates.py

@@ -1582,3 +1582,67 @@ class TestImageAssociationWithTemplate:
 
         response = await client.post(app.url_path_for("create_template"), json=params)
         assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    @pytest.mark.parametrize("reference", ["Vendor/router.bin", "router.bin"])
+    async def test_nested_template_resolves_image_in_correct_type_root(self, client, images_dir, db_session, reference):
+        repository = ImagesRepository(db_session)
+        # Insert the wrong type first, then a suffix collision below QEMU itself.
+        for index, relative in enumerate([f"IOU/{reference}", f"QEMU/Other/{reference}", f"QEMU/{reference}"]):
+            path = Path(images_dir) / relative
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(b"image bytes")
+            await repository.add_image(
+                path.name, "iou" if index == 0 else "qemu", 11, str(path), str(index) * 32, "md5"
+            )
+        expected_path = str(Path(images_dir) / "QEMU" / reference)
+        response = await client.post(
+            "/v3/templates",
+            json={
+                "name": f"Nested QEMU {reference}",
+                "template_type": "qemu",
+                "compute_id": "local",
+                "hda_disk_image": reference,
+                "ram": 512,
+            },
+        )
+        assert response.status_code == 201, response.text
+        templates = TemplatesRepository(db_session)
+        template = await templates.get_template(uuid.UUID(response.json()["template_id"]))
+        assert [image.path for image in template.images] == [expected_path]
+        # Updating removes the association for the same correctly resolved path.
+        response = await client.put(f"/v3/templates/{template.template_id}", json={"hda_disk_image": ""})
+        assert response.status_code == 200, response.text
+        await db_session.refresh(template, ["images"])
+        assert template.images == []
+
+    async def test_update_removes_legacy_association_when_new_exact_path_exists(self, client, images_dir, db_session):
+        repository = ImagesRepository(db_session)
+        legacy_path = Path(images_dir) / "QEMU/Other/Vendor/router.qcow2"
+        canonical_path = Path(images_dir) / "QEMU/Vendor/router.qcow2"
+        for index, path in enumerate([legacy_path, canonical_path]):
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(b"image bytes")
+            await repository.add_image(path.name, "qemu", 11, str(path), str(index) * 32, "md5")
+        response = await client.post(
+            "/v3/templates",
+            json={
+                "name": "Legacy nested association",
+                "template_type": "qemu",
+                "compute_id": "local",
+                "hda_disk_image": str(legacy_path),
+                "ram": 512,
+            },
+        )
+        assert response.status_code == 201, response.text
+        templates = TemplatesRepository(db_session)
+        template = await templates.get_template(uuid.UUID(response.json()["template_id"]))
+        # Simulate an existing template that used the legacy suffix lookup.
+        await templates.update_template(template, {"hda_disk_image": "Vendor/router.qcow2"})
+        legacy_image = template.images[0]
+        legacy_image.availability = "missing"
+        await db_session.commit()
+        legacy_path.unlink()
+        response = await client.put(f"/v3/templates/{template.template_id}", json={"hda_disk_image": ""})
+        assert response.status_code == 200, response.text
+        await db_session.refresh(template, ["images"])
+        assert template.images == []

tests/controller/test_appliance_manager.py

@@ -15,7 +15,12 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import copy
+import json
 import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -160,3 +165,67 @@ async def test_install_version_not_found(monkeypatch):
 
     with pytest.raises(ControllerNotFoundError):
         await manager.install_appliance(uuid.UUID(appliance.id), "9.9", None, None, None, None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "dependency_state",
+    ["missing", "selected", "existing", "downloadable"],
+)
+async def test_image_compatibility_accounts_for_dependencies_without_installing(dependency_state):
+    data = json.loads(Path("gns3server/appliances/empty-vm.gns3a").read_text())
+    primary, dependency = data["images"][:2]
+    if dependency_state != "downloadable":
+        dependency.pop("direct_download_url")
+    data["images"] = [primary, dependency]
+    data["versions"] = [
+        {
+            "name": primary["version"],
+            "images": {"hda_disk_image": primary["filename"], "hdb_disk_image": dependency["filename"]},
+        }
+    ]
+    original = copy.deepcopy(data)
+    appliance = Appliance("test.gns3a", data)
+    manager = ApplianceManager()
+    manager._appliances[appliance.id] = appliance
+    repo = AsyncMock()
+    repo.get_image_by_checksum.return_value = object() if dependency_state == "existing" else None
+    manager._download_image = AsyncMock()
+    manager._create_template = AsyncMock()
+    checksums = [primary["md5sum"]]
+    if dependency_state == "selected":
+        checksums.append(dependency["md5sum"])
+    result = await manager.check_image_compatibility(checksums, repo)
+    match = result[0]["matches"][0]
+    assert match["name"] == "Empty VM"
+    assert match["version"] == primary["version"]
+    assert match["missing_images"] == ([dependency["filename"]] if dependency_state == "missing" else [])
+    assert match["downloadable_images"] == ([dependency["filename"]] if dependency_state == "downloadable" else [])
+    manager._download_image.assert_not_awaited()
+    manager._create_template.assert_not_awaited()
+    assert appliance._data == original
+
+
+@pytest.mark.asyncio
+async def test_image_compatibility_returns_no_matches_for_custom_checksum():
+    manager = ApplianceManager()
+    repo = AsyncMock()
+    result = await manager.check_image_compatibility(["A" * 32, "a" * 32], repo)
+    assert result == [{"checksum": "a" * 32, "matches": []}]
+    repo.get_image_by_checksum.assert_not_awaited()
+
+
+def test_image_catalog_filter_is_conservative_and_does_not_match_by_filename():
+    manager = ApplianceManager()
+    manager._appliances = {
+        "a": SimpleNamespace(
+            images=[
+                {"md5sum": "a" * 32, "filesize": 123, "filename": "router.qcow2"},
+                {"md5sum": "b" * 32, "filesize": 123},
+                {"filesize": 456},
+            ]
+        )
+    }
+    assert manager.image_compatibility_catalog() == {"image_sizes": [123], "has_unknown_sizes": False}
+    manager._appliances["a"].images.append({"md5sum": "c" * 32})
+    assert manager.image_compatibility_catalog()["has_unknown_sizes"] is True

tests/services/test_image_reconciliation.py

@@ -707,3 +707,131 @@ async def test_delete_preserves_replacement_created_while_waiting_for_lock(inven
                 assert await repository.prune_images() == 0
     assert path.read_bytes() == replacement
     assert (await rows(inventory))[0]["image_id"] == replacement_id
+
+
+async def upload_stream(inventory, config, stream, *, subdirectory, filename="router.qcow2"):
+    async with AsyncSession(inventory.engine) as session:
+        return await write_image(
+            filename,
+            str(Path(config.settings.Server.images_path) / filename),
+            stream,
+            ImagesRepository(session),
+            subdirectory=subdirectory,
+        )
+
+
+async def test_upload_rejects_directory_swap_and_cleans_anchored_temporary(inventory, config, tmp_path):
+    if os.open not in os.supports_dir_fd:
+        pytest.skip("Requires directory descriptor support")
+    root = Path(config.settings.Server.images_path)
+    folder = root / "QEMU" / "vendor"
+    moved = root / "QEMU" / "original-vendor"
+    outside = tmp_path / "outside"
+    outside.mkdir()
+
+    async def swapping_stream():
+        yield QCOW
+        folder.rename(moved)
+        folder.symlink_to(outside, target_is_directory=True)
+        yield b"more bytes"
+
+    with pytest.raises(OSError):
+        await upload_stream(inventory, config, swapping_stream(), subdirectory="vendor")
+    assert not list(outside.iterdir())
+    assert not list(moved.iterdir())
+    assert not await rows(inventory)
+
+
+async def test_cancelled_nested_upload_cleans_temporary(inventory, config):
+    root = Path(config.settings.Server.images_path)
+
+    async def cancelled_stream():
+        yield QCOW
+        raise asyncio.CancelledError
+
+    with pytest.raises(asyncio.CancelledError):
+        await upload_stream(inventory, config, cancelled_stream(), subdirectory="Vendor/Version")
+    assert not list((root / "QEMU/Vendor/Version").iterdir())
+    assert not await rows(inventory)
+
+
+async def test_nested_appliance_reference_selects_correct_same_named_image(inventory, config):
+    from gns3server.controller.appliance_manager import ApplianceManager
+    from gns3server.db.repositories.templates import TemplatesRepository
+
+    image_file(config, "QEMU/other/router.qcow2", QCOW + b"other")
+    path = image_file(config, "QEMU/vendor/router.qcow2", QCOW + b"required")
+    await scan(inventory)
+    checksum = hashlib.md5(path.read_bytes(), usedforsecurity=False).hexdigest()
+    appliance = SimpleNamespace(images=[{"filename": path.name, "md5sum": checksum}])
+    version = {"images": {"hda_disk_image": path.name}}
+    async with AsyncSession(inventory.engine, expire_on_commit=False) as session:
+        await ApplianceManager()._find_appliance_version_images(
+            appliance, version, ImagesRepository(session), str(path.parent)
+        )
+        reference = version["images"]["hda_disk_image"]
+        assert reference == "vendor/router.qcow2"
+        resolved = await TemplatesRepository(session).get_image(reference)
+        assert resolved.path == str(path)
+        assert resolved.checksum == checksum
+
+
+@pytest.mark.parametrize("anchored", [True, False])
+async def test_upload_directory_publication_is_no_overwrite_in_both_modes(config, anchored):
+    from gns3server.utils.image_inventory import ImageUploadDirectory
+
+    root = Path(config.settings.Server.images_path)
+    directory = ImageUploadDirectory(str(root), str(root / "QEMU/Vendor/Version"))
+    if anchored and not directory.anchored:
+        pytest.skip("Requires directory descriptor support")
+    directory.anchored = anchored
+    with directory:
+        fd, temporary = directory.temporary()
+        os.write(fd, QCOW)
+        os.close(fd)
+        directory.publish(temporary, "router.qcow2")
+        fd, temporary = directory.temporary()
+        os.write(fd, QCOW + b"replacement")
+        os.close(fd)
+        with pytest.raises(FileExistsError):
+            directory.publish(temporary, "router.qcow2")
+        directory.remove(temporary)
+    assert (root / "QEMU/Vendor/Version/router.qcow2").read_bytes() == QCOW
+
+
+@pytest.mark.parametrize("anchored", [True, False])
+@pytest.mark.parametrize("inside", [True, False])
+async def test_upload_directory_rejects_symlink_before_creating_children(config, tmp_path, anchored, inside):
+    from gns3server.utils.image_inventory import ImageUploadDirectory
+
+    root = Path(config.settings.Server.images_path)
+    (root / "QEMU").mkdir(parents=True, exist_ok=True)
+    target = root / "target" if inside else tmp_path / "outside"
+    target.mkdir()
+    (root / "QEMU/link").symlink_to(target, target_is_directory=True)
+    directory = ImageUploadDirectory(str(root), str(root / "QEMU/link/child"))
+    if anchored and not directory.anchored:
+        pytest.skip("Requires directory descriptor support")
+    directory.anchored = anchored
+    with pytest.raises(OSError), directory:
+        pytest.fail("A symlink folder must not be accepted")
+    assert not list(target.iterdir())
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits and umask")
+async def test_upload_restores_executable_permissions_under_restrictive_umask(inventory, config):
+    import stat
+
+    root = Path(config.settings.Server.images_path)
+    (root / "IOU/Vendor").mkdir(parents=True)
+    (Path(config.config_dir) / ".image-locks").mkdir(parents=True, exist_ok=True)
+
+    async def stream():
+        yield b"\x7fELF\x02\x01\x01"
+
+    previous_umask = os.umask(0o777)
+    try:
+        image = await upload_stream(inventory, config, stream(), filename="router.bin", subdirectory="Vendor")
+    finally:
+        os.umask(previous_umask)
+    assert stat.S_IMODE(os.stat(image.path).st_mode) == 0o700