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