Merge pull request #2840 from GNS3/fix-unvalidated-symlink

Fix unvalidated symlink creation in import_project

gns3server/controller/import_project.py

@@ -166,24 +166,52 @@ async def import_project(
     project = await controller.load_project(dot_gns3_path, load=False)
     return project
 
+
 def _create_symbolic_links(zip_file, path):
     """
     Manually create symbolic links (if any) because ZipFile does not support it.
+    Refuse any target that escapes `path`.
 
     :param zip_file: ZipFile instance
     :param path: project location
     """
 
+    path_root = os.path.realpath(path) + os.sep
     for zip_info in zip_file.infolist():
-        if stat.S_ISLNK(zip_info.external_attr >> 16):
-            symlink_target = zip_file.read(zip_info.filename).decode()
-            symlink_path = os.path.join(path, zip_info.filename)
-            try:
-                # remove the regular file and replace it by a symbolic link
-                os.remove(symlink_path)
-                os.symlink(symlink_target, symlink_path)
-            except OSError as e:
-                raise aiohttp.web.HTTPConflict(text=f"Cannot create symbolic link: {e}")
+        if not stat.S_ISLNK(zip_info.external_attr >> 16):
+            continue
+        symlink_target = zip_file.read(zip_info.filename).decode()
+        symlink_path = os.path.join(path, zip_info.filename)
+
+        # 1. Reject absolute targets outright.
+        if os.path.isabs(symlink_target):
+            raise aiohttp.web.HTTPConflict(
+                text=f"Symlink {zip_info.filename!r} has absolute target {symlink_target!r}, refusing"
+            )
+
+        # 2. Reject paths where the entry name itself escapes (defence in depth;
+        #    extractall normally would already have caught this).
+        member_abs = os.path.realpath(symlink_path)
+        if not (member_abs + os.sep).startswith(path_root) and member_abs + os.sep != path_root:
+            raise aiohttp.web.HTTPConflict(
+                text=f"Symlink entry {zip_info.filename!r} escapes project dir, refusing"
+            )
+
+        # 3. Resolve the symlink target relative to the entry's own parent
+        #    directory and verify the resolved real path stays inside `path`.
+        link_dir = os.path.realpath(os.path.dirname(symlink_path))
+        resolved_target = os.path.realpath(os.path.join(link_dir, symlink_target))
+        if not (resolved_target + os.sep).startswith(path_root) and resolved_target + os.sep != path_root:
+            raise aiohttp.web.HTTPConflict(
+                text=f"Symlink {zip_info.filename!r} -> {symlink_target!r} escapes project dir, refusing"
+            )
+
+        try:
+            os.remove(symlink_path)
+            os.symlink(symlink_target, symlink_path)
+        except OSError as e:
+            raise aiohttp.web.HTTPConflict(text=f"Cannot create symbolic link: {e}")
+
 
 def regenerate_topology_ids(topology, new_project_path, reset_mac_addresses=False):
     """

tests/controller/test_import_project.py

@@ -19,6 +19,8 @@ import os
 import uuid
 import json
 import zipfile
+import pytest
+import aiohttp
 
 from tests.utils import asyncio_patch, AsyncioMagicMock
 from unittest.mock import patch, MagicMock
@@ -117,36 +119,48 @@ async def write_file(path, z):
             f.write(chunk)
 
 
-async def test_import_project_containing_symlink(tmpdir, controller):
+@pytest.fixture
+def export_project_with_symlink(tmpdir, controller):
+    async def _export(symlink_target):
+        project = Project(controller=controller, name="test")
+        project.dump = MagicMock()
 
-    project = Project(controller=controller, name="test")
-    project.dump = MagicMock()
-    path = project.path
+        topology = {
+            "project_id": str(uuid.uuid4()),
+            "name": "test",
+            "auto_open": True,
+            "auto_start": True,
+            "topology": {
+            },
+            "version": "2.0.0"
+        }
+
+        with open(os.path.join(project.path, "project.gns3"), 'w+') as f:
+            json.dump(topology, f)
+
+        os.makedirs(os.path.join(project.path, "vm1", "dynamips"))
+        symlink_path = os.path.join(project.path, "vm1", "dynamips", "symlink")
+        os.symlink(symlink_target, symlink_path)
+
+        zip_path = str(tmpdir / "project.zip")
+        with aiozipstream.ZipFile() as z:
+            with patch("gns3server.compute.Dynamips.get_images_directory", return_value=str(tmpdir / "IOS"),):
+                await export_project(z, project, str(tmpdir), include_images=False)
+                await write_file(zip_path, z)
+
+        return zip_path
+
+    return _export
+
+
+async def test_import_project_containing_symlink(controller, export_project_with_symlink):
+    """
+    Test importing a project containing a valid symlink (target inside the project directory).
+    """
 
     project_id = str(uuid.uuid4())
-    topology = {
-        "project_id": str(uuid.uuid4()),
-        "name": "test",
-        "auto_open": True,
-        "auto_start": True,
-        "topology": {
-        },
-        "version": "2.0.0"
-    }
-
-    with open(os.path.join(path, "project.gns3"), 'w+') as f:
-        json.dump(topology, f)
-
-    os.makedirs(os.path.join(path, "vm1", "dynamips"))
-    symlink_path = os.path.join(project.path, "vm1", "dynamips", "symlink")
-    symlink_target = "/tmp/anywhere"
-    os.symlink(symlink_target, symlink_path)
-
-    zip_path = str(tmpdir / "project.zip")
-    with aiozipstream.ZipFile() as z:
-        with patch("gns3server.compute.Dynamips.get_images_directory", return_value=str(tmpdir / "IOS"),):
-            await export_project(z, project, str(tmpdir), include_images=False)
-            await write_file(zip_path, z)
+    symlink_target = "../symlink_target"
+    zip_path = await export_project_with_symlink(symlink_target)
 
     with open(zip_path, "rb") as f:
         project = await import_project(controller, project_id, f)
@@ -158,6 +172,36 @@ async def test_import_project_containing_symlink(tmpdir, controller):
     assert os.readlink(symlink_path) == symlink_target
 
 
+async def test_import_project_containing_absolute_symlink(controller, export_project_with_symlink):
+    """
+    Test importing a project containing an absolute symlink.
+    This should fail because absolute symlinks are not allowed for security reasons.
+    """
+
+    project_id = str(uuid.uuid4())
+    symlink_target = "/tmp/anywhere"
+    zip_path = await export_project_with_symlink(symlink_target)
+
+    with pytest.raises(aiohttp.web.HTTPConflict):
+        with open(zip_path, "rb") as f:
+            await import_project(controller, project_id, f)
+
+
+async def test_import_project_containing_escaping_symlink(controller, export_project_with_symlink):
+    """
+    Test importing a project containing a symlink that escapes the project directory.
+    This should fail because symlinks that escape the project directory are not allowed for security reasons.
+    """
+
+    project_id = str(uuid.uuid4())
+    symlink_target = "../../../../symlink_target"
+    zip_path = await export_project_with_symlink(symlink_target)
+
+    with pytest.raises(aiohttp.web.HTTPConflict):
+        with open(zip_path, "rb") as f:
+            await import_project(controller, project_id, f)
+
+
 async def test_import_upgrade(tmpdir, controller):
     """
     Topology made for previous GNS3 version are upgraded during the process