fix: add safe_extract_zip to prevent Zip Slip (#2812)

Author: xogas

apiserver/paasng/paasng/platform/sourcectl/controllers/github.py

@@ -20,7 +20,6 @@
 from os import PathLike, path, walk
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
-from zipfile import ZipFile
 
 import arrow
 from django.utils.functional import cached_property
@@ -39,6 +38,7 @@
 from paasng.platform.sourcectl.repo_controller import BaseGitRepoController
 from paasng.platform.sourcectl.source_types import get_sourcectl_names
 from paasng.platform.sourcectl.utils import generate_temp_file
+from paasng.utils.archive import safe_extract_zip
 
 logger = logging.getLogger(__name__)
 
@@ -106,7 +106,7 @@ def export(self, local_path: PathLike, version_info: VersionInfo | None = None,
 
         with generate_temp_file(suffix=".zip") as zip_file:
             self.api_client.repo_archive(self.project, zip_file, ref=ref)
-            ZipFile(zip_file, "r").extractall(local_path)
+            safe_extract_zip(zip_file, local_path)
 
         # Github 下载的 zip 包比较特殊，外层有格式为 {username}-{proj_name}-{ref} 的目录，需要平铺开
         for root, dirs, _ in walk(str(local_path)):

apiserver/paasng/paasng/platform/sourcectl/package/client.py

@@ -35,6 +35,7 @@
 from paasng.platform.sourcectl.models import SourcePackage
 from paasng.platform.sourcectl.package.downloader import download_file_via_url
 from paasng.platform.sourcectl.utils import generate_temp_dir, uncompress_directory
+from paasng.utils.archive import safe_extract_zip
 from paasng.utils.text import remove_prefix
 
 logger = logging.getLogger(__name__)
@@ -269,11 +270,9 @@ def read_file(self, file_path: str) -> bytes:
         return self.zip_.read(info)
 
     def export(self, local_path: str):
-        self.zip_.extractall(local_path)  # noqa: S202
-
-        # About symbolic links security check, the zip file module does not support symbolic links
-        # at this moment so no extra checking is needed.
-        # See https://bugs.python.org/issue37921 for more details
+        # 使用 safe_extract_zip 来解压 zip 包, 以防止 Zip Slip 漏洞,
+        # safe_extract_zip 会对每个 zip 成员进行路径校验, 如果发现不安全的成员路径会抛出 UnsafeArchiveError 异常
+        safe_extract_zip(self.zip_, local_path)
 
     def close(self):
         self.zip_.close()

apiserver/paasng/paasng/utils/archive.py

@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+# TencentBlueKing is pleased to support the open source community by making
+# 蓝鲸智云 - PaaS 平台 (BlueKing - PaaS System) available.
+# Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
+# Licensed under the MIT License (the "License"); you may not use this file except
+# in compliance with the License. You may obtain a copy of the License at
+#
+#     http://opensource.org/licenses/MIT
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+# either express or implied. See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# We undertake not to change the open source license (MIT license) applicable
+# to the current version of the project delivered to anyone in the future.
+
+"""安全解压归档文件的工具
+
+提供 `safe_extract_zip` 函数, 用于安全地解压 ZIP 文件, 防止 Zip Slip (CWE-22) 漏洞.
+"""
+
+import contextlib
+import logging
+import os
+from pathlib import Path
+from typing import Union
+from zipfile import ZipFile, ZipInfo
+
+logger = logging.getLogger(__name__)
+
+
+class UnsafeArchiveError(Exception):
+    """当归档文件中包含可能逃逸目标目录的成员时抛出.
+
+    包括但不限于: 绝对路径, 包含 `..` 的相对路径穿越,
+    解析后落在目标目录之外的路径, 指向目录外部的符号链接等
+    """
+
+
+def safe_extract_zip(zip_file: Union[str, Path, ZipFile], local_path: Union[str, os.PathLike]) -> None:
+    """安全解压 zip 归档文件到指定目录, 在写出每个成员之前进行路径检查, 防止 Zip Slip 漏洞.
+
+    :param zip_file: ZIP 文件路径或已打开的 ZipFile 对象
+    :param local_path: 解压目标目录
+    :raises UnsafeArchiveError: 如果归档文件中包含不安全的成员路径
+    """
+    base = Path(local_path).resolve()
+    base.mkdir(parents=True, exist_ok=True)
+
+    with contextlib.ExitStack() as stack:
+        if isinstance(zip_file, ZipFile):
+            zf = zip_file
+        else:
+            zf = stack.enter_context(ZipFile(str(zip_file), "r"))
+
+        members = zf.infolist()
+        for info in members:
+            _validate_zip_info(info, base)
+        # 所有成员校验通过, 执行解压
+        for info in members:
+            zf.extract(info, path=str(base))
+
+
+def _validate_zip_info(info: ZipInfo, base: Path) -> None:
+    """校验单个 zip 成员的安全性
+
+    :param info: ZipInfo 对象
+    :param base: 解压目录的绝对路径 (已 resolve)
+    :raises UnsafeArchiveError: 如果成员路径不安全
+    """
+    # 统一路径分隔符为 "/"
+    name = info.filename.replace("\\", "/")
+
+    # 拒绝空成员名
+    if not name or name in (".", "./"):
+        logger.warning("Zip archive contains an empty member name, extraction denied")
+        raise UnsafeArchiveError("Zip member has an empty name, which is not allowed")
+
+    # 拒绝绝对路径
+    if os.path.isabs(name):
+        logger.warning("Zip archive contains a member with absolute path: %s, extraction denied", name)
+        raise UnsafeArchiveError(f"Zip member '{name}' has an absolute path, which is not allowed")
+
+    # 归一化后校验最终路径是否在 base 目录下
+    target_path = os.path.realpath(os.path.join(base, name))
+    if os.path.commonpath([base, target_path]) != str(base):
+        logger.warning(
+            "Zip archive contains a path traversal member: %s (resolves to %s), extraction denied", name, target_path
+        )
+        raise UnsafeArchiveError(
+            f"Zip member '{name}' resolves to '{target_path}', which is outside the target directory '{base}'"
+        )

apiserver/paasng/tests/paasng/test_utils/test_archive.py

@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+# TencentBlueKing is pleased to support the open source community by making
+# 蓝鲸智云 - PaaS 平台 (BlueKing - PaaS System) available.
+# Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
+# Licensed under the MIT License (the "License"); you may not use this file except
+# in compliance with the License. You may obtain a copy of the License at
+#
+#     http://opensource.org/licenses/MIT
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+# either express or implied. See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# We undertake not to change the open source license (MIT license) applicable
+# to the current version of the project delivered to anyone in the future.
+
+import zipfile
+from pathlib import Path
+
+import pytest
+
+from paasng.utils.archive import UnsafeArchiveError, safe_extract_zip
+
+
+class TestSafeExtractZip:
+    @pytest.fixture
+    def dest_dir(self, tmp_path: Path) -> Path:
+        """解压目标目录"""
+        return tmp_path / "dest"
+
+    @staticmethod
+    def _make_zip(zip_path: Path, members: dict) -> Path:
+        """根据 {成员名: 内容} 创建一个 zip 文件, 内容为 None 时表示目录"""
+        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+            for name, content in members.items():
+                if content is None:
+                    # 目录成员, 以 "/" 结尾
+                    zi = zipfile.ZipInfo(name if name.endswith("/") else name + "/")
+                    zf.writestr(zi, b"")
+                else:
+                    zf.writestr(name, content)
+        return zip_path
+
+    def test_extract_valid_zip(self, tmp_path: Path, dest_dir: Path):
+        """测试正常的 zip 文件能够正确解压"""
+        members = {
+            "file1.txt": b"Hello, World!",
+            "./file2.txt": b"File with dot",
+            "dir/": None,
+            "dir/file3.txt": b"Another file",
+            "dir/./file4.txt": b"File with dot in dir",
+        }
+        zip_path = self._make_zip(tmp_path / "test.zip", members)
+        safe_extract_zip(zip_path, dest_dir)
+
+        # 验证解压后的文件内容
+        assert (dest_dir / "file1.txt").read_bytes() == b"Hello, World!"
+        assert (dest_dir / "file2.txt").read_bytes() == b"File with dot"
+        assert (dest_dir / "dir").is_dir()
+        assert (dest_dir / "dir/file3.txt").read_bytes() == b"Another file"
+        assert (dest_dir / "dir/file4.txt").read_bytes() == b"File with dot in dir"
+
+    def test_reject_absolute_path_member(self, tmp_path: Path, dest_dir: Path):
+        """测试包含绝对路径成员的 zip 文件会被拒绝"""
+        members = {
+            "/absolute.txt": b"Should not be extracted",
+        }
+        zip_path = self._make_zip(tmp_path / "test.zip", members)
+        with pytest.raises(UnsafeArchiveError):
+            safe_extract_zip(zip_path, dest_dir)
+
+    def test_reject_parent_traversal_member(self, tmp_path: Path, dest_dir: Path):
+        """测试包含路径穿越成员的 zip 文件会被拒绝"""
+        members = {
+            "../evil1.txt": b"Should not be extracted",
+        }
+        zip_path = self._make_zip(tmp_path / "test.zip", members)
+        with pytest.raises(UnsafeArchiveError):
+            safe_extract_zip(zip_path, dest_dir)
+
+    def test_reject_complex_traversal_member(self, tmp_path: Path, dest_dir: Path):
+        """测试包含复杂路径穿越成员的 zip 文件会被拒绝"""
+        members = {
+            "dir/../../evil2.txt": b"Should not be extracted",
+        }
+        zip_path = self._make_zip(tmp_path / "test.zip", members)
+        with pytest.raises(UnsafeArchiveError):
+            safe_extract_zip(zip_path, dest_dir)
+
+    def test_no_partial_extraction_on_unsafe_member(self, tmp_path: Path, dest_dir: Path):
+        """测试当 zip 文件中包含不安全成员时，不会解压任何成员"""
+        members = {
+            "good.txt": b"Good file",
+            "../evil.txt": b"Should not be extracted",
+        }
+        zip_path = self._make_zip(tmp_path / "test.zip", members)
+        with pytest.raises(UnsafeArchiveError):
+            safe_extract_zip(zip_path, dest_dir)
+
+        # dest_dir 应该保持为空, 没有任何成员被解压
+        assert dest_dir.is_dir()
+        assert not any(dest_dir.iterdir())
+        # 父目录中也不应该有 evil.txt
+        assert not (dest_dir.parent / "evil.txt").exists()