"""覆盖原生目录入口的归档与 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()