Blame for tests/native_api/test_catalog.py

"""覆盖原生目录入口的归档与 npm manifest 信任边界。"""

import gzip
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import tarfile
import unittest
from unittest.mock import patch


ROOT = Path(__file__).resolve().parents[2]
SPEC = importlib.util.spec_from_file_location("native_catalog_under_test", ROOT / "apps/market-api/app/native/catalog.py")
catalog = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(catalog)


def manifest():
    return {
        "name": "@message/example",
        "version": "1.2.3-rc.2+build.4",
        "description": "原生测试插件",
        "scripts": {"test": "node --test", "build": "tsc"},
        "dependencies": {"example-dependency": "2.3.4"},
        "peerDependencies": {"@deepseek-ai/dsh-tools": "0.2.0-rc.2"},
        "dsh": {"manifestVersion": 1, "bundle": {"patch": "./cordis.patch.yml"}},
    }


def archive(entries, *, format=tarfile.PAX_FORMAT):
    stream = io.BytesIO()
    with tarfile.open(fileobj=stream, mode="w", format=format) as handle:
        for entry in entries:
            name, payload, *options = entry
            info = tarfile.TarInfo(name)
            info.size = len(payload)
            if options:
                info.type = options[0]
            if len(options) > 1:
                info.linkname = options[1]
            handle.addfile(info, io.BytesIO(payload))
    return gzip.compress(stream.getvalue(), mtime=0)


def bundle(metadata=None, extras=()):
    if metadata is None:
        metadata = manifest()
    return archive([
        ("package/package.json", json.dumps(metadata, ensure_ascii=False).encode()),
        ("package/cordis.patch.yml", b"- name: example\n"),
        *extras,
    ])


class CatalogArchiveTests(unittest.TestCase):
    def assert_rejected(self, data, message=None):
        with self.assertRaises(ValueError) as caught:
            catalog.validate_archive(data)
        # 接口直接展示校验失败的原因，所有错误均应使用中文。
        self.assertRegex(str(caught.exception), r"[\u4e00-\u9fff]")
        if message:
            self.assertIn(message, str(caught.exception))

    def test_valid_bundle_returns_original_archive_digest(self):
        data = bundle()
        self.assertEqual(catalog.validate_archive(data), {
            "name": "@message/example", "version": "1.2.3-rc.2+build.4",
            "description": "原生测试插件", "sha256": hashlib.sha256(data).hexdigest(),
            "bytes": len(data),
        })

    def test_official_bundle_manifests_and_files_are_accepted(self):
        for plugin in ("mcp-export", "text-tools"):
            with self.subTest(plugin=plugin):
                directory = ROOT / "plugins" / plugin
                entries = [("package/" + str(file.relative_to(directory)), file.read_bytes())
                           for file in sorted(directory.rglob("*")) if file.is_file()]
                result = catalog.validate_archive(archive(entries))
                self.assertEqual(result["name"], json.loads((directory / "package.json").read_text())["name"])

    def test_optional_description_and_manifest_version(self):
        metadata = manifest()
        del metadata["description"]
        del metadata["dsh"]["manifestVersion"]
        metadata["dsh"]["bundle"]["patch"] = "cordis.patch.yml"
        self.assertEqual(catalog.validate_archive(bundle(metadata))["description"], "")

    def test_directories_unicode_and_long_paths_are_supported(self):
        self.assertTrue(catalog.validate_archive(bundle(extras=[
            ("package/", b"", tarfile.DIRTYPE),
            ("package/src/", b"", tarfile.DIRTYPE),
            ("package/src/工具.mjs", b"export {}"),
            ("package/src/" + "a" * 200 + ".mjs", b"export {}"),
        ])))

    def test_gnu_long_paths_are_supported(self):
        data = archive([
            ("package/package.json", json.dumps(manifest()).encode()),
            ("package/cordis.patch.yml", b"[]"),
            ("package/" + "a" * 200, b"test"),
        ], format=tarfile.GNU_FORMAT)
        self.assertTrue(catalog.validate_archive(data))

    def test_empty_non_bytes_and_non_gzip_input(self):
        for data in (b"", "test", None, bytearray(b"test"), b"not a tar", bytes(2048)):
            with self.subTest(data=repr(data)[:30]):
                self.assert_rejected(data)

    def test_upload_limit_and_invalid_limit_configuration(self):
        data = bundle()
        self.assertTrue(catalog.validate_archive(data, len(data)))
        with self.assertRaises(ValueError):
            catalog.validate_archive(data, len(data) - 1)
        for value in (0, -1, True, "100"):
            with self.subTest(value=value), self.assertRaises(ValueError):
                catalog.validate_archive(data, value)

    def test_corrupt_and_truncated_gzip_are_rejected(self):
        data = bundle()
        corrupted = bytearray(data)
        corrupted[-8] ^= 255
        for value in (data[:5], data[:-5], bytes(corrupted), data + b"garbage"):
            with self.subTest(size=len(value)):
                self.assert_rejected(value)

    def test_invalid_tar_checksum_and_truncated_tar(self):
        raw = gzip.decompress(bundle())
        corrupt = bytearray(raw)
        corrupt[0] ^= 32
        for value in (bytes(corrupt), raw[:513], raw[:1024], b"x" * 1024):
            with self.subTest(size=len(value)):
                self.assert_rejected(gzip.compress(value))

    def test_missing_manifest_and_manifest_directory(self):
        self.assert_rejected(archive([("package/main.mjs", b"export {}")]), "package/package.json")
        self.assert_rejected(archive([("package/package.json", b"", tarfile.DIRTYPE)]))

    def test_unsafe_paths_are_rejected(self):
        for name in (
            "/package/x", "../x", "package/../x", "package/a/../../x",
            "package/./x", "package//x", "./package/x", "other/x",
            "package\\x", "package/a\\..\\x", "C:/package/x",
            "package/C:x", "package/x\ny", "package/x\x7fy", "package",
        ):
            with self.subTest(name=name):
                self.assert_rejected(bundle(extras=[(name, b"evil")]))

    def test_noncanonical_directory_paths_are_rejected(self):
        for name in ("package//", "package/./", "package/a//", "package/../"):
            with self.subTest(name=name):
                self.assert_rejected(bundle(extras=[(name, b"", tarfile.DIRTYPE)]))

    def test_duplicate_and_conflicting_paths_are_rejected(self):
        self.assert_rejected(bundle(extras=[("package/package.json", b"{}")]), "重复")
        for entries in (
            [("package/a", b"file"), ("package/a/b", b"child")],
            [("package/a/b", b"child"), ("package/a", b"file")],
            [("package/src/", b"", tarfile.DIRTYPE), ("package/src", b"", tarfile.DIRTYPE)],
        ):
            with self.subTest(entries=entries):
                self.assert_rejected(bundle(extras=entries))

    def test_links_devices_fifo_sparse_and_unknown_types_are_rejected(self):
        for kind in (tarfile.SYMTYPE, tarfile.LNKTYPE, tarfile.CHRTYPE, tarfile.BLKTYPE,
                     tarfile.FIFOTYPE, tarfile.GNUTYPE_SPARSE, tarfile.CONTTYPE, b"Z"):
            with self.subTest(kind=kind):
                self.assert_rejected(bundle(extras=[("package/evil", b"", kind, "package/package.json")]))

    def test_regular_files_cannot_smuggle_link_targets(self):
        self.assert_rejected(bundle(extras=[("package/evil", b"", tarfile.REGTYPE, "/tmp/evil")]))

    def test_hidden_second_archive_and_nonzero_padding_are_rejected(self):
        raw = gzip.decompress(bundle())
        self.assert_rejected(gzip.compress(raw + gzip.decompress(bundle())))
        modified = bytearray(raw)
        size = int(modified[124:136].rstrip(b"\0 "), 8)
        modified[512 + size] = 1
        self.assert_rejected(gzip.compress(modified))

    def test_file_count_limit(self):
        data = bundle(extras=[(f"package/f{index}", b"") for index in range(catalog.MAX_MEMBERS - 2)])
        self.assertTrue(catalog.validate_archive(data))
        self.assert_rejected(bundle(extras=[(f"package/f{index}", b"") for index in range(catalog.MAX_MEMBERS - 1)]), "4096")

    def test_file_and_manifest_size_limits(self):
        with patch.object(catalog, "MAX_FILE_BYTES", 1024):
            self.assert_rejected(bundle(extras=[("package/large", b"a" * 1025)]), "单文件")
        metadata = manifest()
        metadata["description"] = "a" * catalog.MAX_MANIFEST_BYTES
        self.assert_rejected(bundle(metadata), "1 MiB")

    def test_real_compression_bomb_is_rejected(self):
        data = gzip.compress(bytes(catalog.MAX_EXPANDED_BYTES + 1))
        self.assertLess(len(data), 100000)
        self.assert_rejected(data, "64 MiB")

    def test_total_expansion_limit_counts_all_files(self):
        data = bundle(extras=[("package/a", b"a" * 4096), ("package/b", b"b" * 4096)])
        with patch.object(catalog, "MAX_EXPANDED_BYTES", 8192):
            self.assert_rejected(data, "64 MiB")

    def test_lockfiles_manager_configs_and_bundled_modules_are_rejected(self):
        for name in (*catalog.FORBIDDEN_FILES, "node_modules/x/index.js", "nested/node_modules/x.js", "NODE_MODULES/a"):
            with self.subTest(name=name):
                self.assert_rejected(bundle(extras=[("package/" + name, b"{}")] ))

    def test_malformed_manifest_json(self):
        for value in (b"{", b"null", b"[]", b'"name"', b'\xff', b'{"name":"a","name":"b"}', b'{"x":NaN}'):
            with self.subTest(value=value):
                self.assert_rejected(archive([("package/package.json", value)]))

    def test_package_names_and_versions(self):
        for name in (None, 1, "", "../evil", "@scope", "A-name", ".hidden", "_private", "has space", "bad~name", "a" * 215, "node_modules", "favicon.ico", "@Scope/name"):
            with self.subTest(name=name):
                metadata = manifest()
                metadata["name"] = name
                self.assert_rejected(bundle(metadata), "包名")
        for version in (None, 1, "latest", "^1.0.0", "v1.0.0", "1.0", "1.01.0", "1.0.0-01", "1.0.0-", "1.0.0+", " 1.0.0", "1.0.0\n", "9007199254740992.0.0", "1.0.0+" + "a" * 256):
            with self.subTest(version=version):
                metadata = manifest()
                metadata["version"] = version
                self.assert_rejected(bundle(metadata), "semver")

    def test_non_string_description(self):
        metadata = manifest()
        metadata["description"] = {"untrusted": "value"}
        self.assert_rejected(bundle(metadata), "description")

    def test_all_install_publish_lifecycle_scripts_are_rejected(self):
        for script in catalog.LIFECYCLE_SCRIPTS:
            with self.subTest(script=script):
                metadata = manifest()
                metadata["scripts"][script] = ""
                self.assert_rejected(bundle(metadata), "生命周期")
        for scripts in ([], None, {"build": []}):
            with self.subTest(scripts=scripts):
                metadata = manifest()
                metadata["scripts"] = scripts
                self.assert_rejected(bundle(metadata), "scripts")

    def test_dependency_sources_ranges_and_bad_maps_are_rejected(self):
        specifiers = (
            "^1.2.3", "~1.2.3", ">=1.0.0", "1", "1.2", "*", "latest", "1.2.3 || 2.0.0",
            "file:../package", "https://example.test/a.tgz", "git+https://example.test/a.git",
            "github:example/project", "user/repo", "workspace:*", "npm:other@1.2.3", "=1.2.3", None,
        )
        for field in ("dependencies", "devDependencies", "optionalDependencies", "peerDependencies"):
            for value in specifiers:
                with self.subTest(field=field, specifier=value):
                    metadata = manifest()
                    metadata.setdefault(field, {})["example"] = value
                    self.assert_rejected(bundle(metadata), "依赖")
            for dependencies in (None, [], "1.2.3"):
                with self.subTest(field=field, dependencies=dependencies):
                    metadata = manifest()
                    metadata[field] = dependencies
                    self.assert_rejected(bundle(metadata), "依赖")

    def test_dependency_overrides_and_bundling_are_rejected(self):
        for field in ("overrides", "resolutions", "pnpm", "workspaces", "bundledDependencies", "bundleDependencies"):
            with self.subTest(field=field):
                metadata = manifest()
                metadata[field] = {}
                self.assert_rejected(bundle(metadata), "依赖")

    def test_core_peers_are_required_and_version_locked(self):
        for peers in ({}, {"@deepseek-ai/dsh-tools": "0.2.0"},
                      {"@deepseek-ai/dsh-tools": "0.2.0-rc.2", "@deepseek-ai/dsh-core": "0.1.0"}):
            with self.subTest(peers=peers):
                metadata = manifest()
                metadata["peerDependencies"] = peers
                self.assert_rejected(bundle(metadata))

    def test_bundle_patch_must_be_present_and_safe(self):
        for value in (None, {}, [], {"bundle": {}}, {"bundle": []}):
            with self.subTest(dsh=value):
                metadata = manifest()
                metadata["dsh"] = value
                self.assert_rejected(bundle(metadata))
        for value in (None, "", "../patch.yml", "/cordis.patch.yml", "././cordis.patch.yml", "a/../cordis.patch.yml", "a\\patch.yml", "missing.yml"):
            with self.subTest(patch=value):
                metadata = manifest()
                metadata["dsh"]["bundle"]["patch"] = value
                self.assert_rejected(bundle(metadata))
        metadata = manifest()
        metadata["dsh"]["bundle"]["patch"] = "src"
        self.assert_rejected(bundle(metadata, extras=[("package/src", b"", tarfile.DIRTYPE)]))

    def test_unknown_manifest_versions_are_rejected(self):
        for value in (None, 0, 2, "1", True, 1.0, {}):
            with self.subTest(version=value):
                metadata = manifest()
                metadata["dsh"]["manifestVersion"] = value
                self.assert_rejected(bundle(metadata), "manifestVersion")


    def test_pax_traversal_duplicate_paths_and_size_override_are_rejected(self):
        def record(key, value):
            body = (" " + key + "=" + value + "\n").encode()
            size = len(body) + 1
            while size != len(str(size)) + len(body):
                size = len(str(size)) + len(body)
            return str(size).encode() + body

        for payload in (
            record("path", "../escape"), record("path", "/package/escape"),
            record("path", "package/a\\b"), record("path", "package/package.json"),
            record("size", "1000000000000"), record("GNU.sparse.realsize", "1000000000"),
            record("linkpath", "../escape"), record("path", "package/a") + record("path", "package/b"),
            b"invalid", b"0 x=y\n", b"999 path=package/a\n", b"9 path=a",
            b"9" * 5000 + b" path=a\n",
        ):
            with self.subTest(payload=payload[:80]):
                self.assert_rejected(bundle(extras=[
                    ("package/PaxHeader", payload, tarfile.XHDTYPE),
                    ("package/innocent", b"abc"),
                ]))

    def test_dangling_chained_and_global_extension_headers_are_rejected(self):
        for extras in (
            [("package/PaxHeader", b"", tarfile.XHDTYPE)],
            [("package/PaxHeader", b"", tarfile.XHDTYPE), ("package/SecondPax", b"", tarfile.XHDTYPE), ("package/a", b"")],
            [("package/GlobalPax", b"", tarfile.XGLTYPE)],
            [("././@LongLink", b"../escape\0", tarfile.GNUTYPE_LONGNAME), ("package/a", b"")],
        ):
            with self.subTest(extras=extras):
                self.assert_rejected(bundle(extras=extras))

    def test_special_permissions_and_nonempty_directories_are_rejected(self):
        self.assert_rejected(bundle(extras=[("package/dir", b"not a directory", tarfile.DIRTYPE)]))
        raw = bytearray(gzip.decompress(bundle()))
        raw[100:108] = b"0004755\0"
        raw[148:156] = b"        "
        raw[148:156] = f"{sum(raw[:512]):06o}\0 ".encode()
        self.assert_rejected(gzip.compress(raw), "特殊权限")

    def test_unbounded_json_numbers_and_deep_nesting_have_chinese_errors(self):
        for value in (b'{"number":' + b"9" * 5000 + b"}", b"[" * 2000 + b"]" * 2000):
            with self.subTest(prefix=value[:20]):
                self.assert_rejected(archive([("package/package.json", value)]))


    def test_path_lengths_and_depth_are_bounded(self):
        for name in ("package/" + "a/" * 1000 + "file", "package/" + "a" * 256,
                     "package/" + "a" * 4097, "package/" + "深" * 100):
            with self.subTest(length=len(name)):
                self.assert_rejected(bundle(extras=[(name, b"test")]))


if __name__ == "__main__":
    unittest.main()