mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 14:01:15 +02:00
feat(image-manager): Add support for nested image folders and appliance compatibility checks
This commit is contained in:
Cristi committed
1 parent
9f9be983cd
commit
207db770c6
12 files changed
+983
-95
No files matched your search
@@ -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()
|
||||
@@ -1582,3 +1582,57 @@ 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 == []
|
||||
Reference in new issue
Block a user