| 1 | import importlib.util |
| 2 | import stat |
| 3 | import tempfile |
| 4 | import unittest |
| 5 | import zipfile |
| 6 | from pathlib import Path |
| 7 | from unittest.mock import patch |
| 8 | |
| 9 | spec = importlib.util.spec_from_file_location( |
| 10 | "stage_preview_bundle", Path(__file__).with_name("stage-preview-bundle.py") |
| 11 | ) |
| 12 | staging = importlib.util.module_from_spec(spec) |
| 13 | spec.loader.exec_module(staging) |
| 14 | |
| 15 | |
| 16 | class StagePreviewBundleTests(unittest.TestCase): |
| 17 | def setUp(self): |
| 18 | self.temp = tempfile.TemporaryDirectory() |
| 19 | self.addCleanup(self.temp.cleanup) |
| 20 | self.root = Path(self.temp.name) |
| 21 | self.archive = self.root / "bundle.zip" |
| 22 | self.destination = self.root / "staged" |
| 23 | |
| 24 | def bundle(self, extra=(), missing=None): |
| 25 | with zipfile.ZipFile(self.archive, "w") as bundle: |
| 26 | for name in sorted(staging.REQUIRED_FILES - {missing}): |
| 27 | bundle.writestr(name, b"bundle data, never executed") |
| 28 | for name, content in extra: |
| 29 | bundle.writestr(name, content) |
| 30 | |
| 31 | def stage(self): |
| 32 | staging.stage_bundle(self.archive, self.destination) |
| 33 | |
| 34 | def test_preserves_valid_bundle_layout_and_bytes(self): |
| 35 | self.bundle([("server/", b""), ("server/dist/", b""), |
| 36 | ("desktop/dist-electron/chunks/helper.cjs", b"chunk")]) |
| 37 | self.stage() |
| 38 | for name in staging.REQUIRED_FILES: |
| 39 | self.assertEqual((self.destination / name).read_bytes(), b"bundle data, never executed") |
| 40 | self.assertEqual((self.destination / "desktop/dist-electron/chunks/helper.cjs").read_bytes(), b"chunk") |
| 41 | |
| 42 | def test_rejects_builder_overwrite_and_unsafe_paths_before_writing(self): |
| 43 | for name in [ |
| 44 | "desktop/node_modules/electron-builder/cli.js", |
| 45 | "desktop/package.json", |
| 46 | "server/dist/../../desktop/package.json", |
| 47 | "../package.json", |
| 48 | "/server/dist/absolute", |
| 49 | "server/dist/./alias", |
| 50 | "server/dist//alias", |
| 51 | "server/dist/back\\slash", |
| 52 | "server/dist/file:stream", |
| 53 | "server/dist/BIN.MJS", |
| 54 | ]: |
| 55 | with self.subTest(name=name): |
| 56 | self.bundle([(name, b"untrusted")]) |
| 57 | with self.assertRaises(ValueError): |
| 58 | self.stage() |
| 59 | self.assertFalse(self.destination.exists()) |
| 60 | |
| 61 | def test_rejects_links_and_special_files(self): |
| 62 | for mode in [stat.S_IFLNK, stat.S_IFIFO, stat.S_IFCHR]: |
| 63 | with self.subTest(mode=mode): |
| 64 | entry = zipfile.ZipInfo("server/dist/link") |
| 65 | entry.create_system = 3 |
| 66 | entry.external_attr = (mode | 0o777) << 16 |
| 67 | self.bundle([(entry, b"../../../desktop/node_modules")]) |
| 68 | with self.assertRaises(ValueError): |
| 69 | self.stage() |
| 70 | self.assertFalse(self.destination.exists()) |
| 71 | |
| 72 | def test_requires_entry_points(self): |
| 73 | self.bundle(missing="desktop/dist-electron/main.cjs") |
| 74 | with self.assertRaises(ValueError): |
| 75 | self.stage() |
| 76 | self.assertFalse(self.destination.exists()) |
| 77 | |
| 78 | def test_bounds_archive_size_expanded_size_and_entry_count(self): |
| 79 | for limit in ["MAX_ARCHIVE_BYTES", "MAX_EXPANDED_BYTES", "MAX_ENTRIES"]: |
| 80 | with self.subTest(limit=limit), patch.object(staging, limit, 1): |
| 81 | self.bundle() |
| 82 | with self.assertRaises(ValueError): |
| 83 | self.stage() |
| 84 | self.assertFalse(self.destination.exists()) |
| 85 | |
| 86 | def test_refuses_existing_destination(self): |
| 87 | self.bundle() |
| 88 | self.destination.mkdir() |
| 89 | sentinel = self.destination / "trusted" |
| 90 | sentinel.write_text("untouched") |
| 91 | with self.assertRaises(FileExistsError): |
| 92 | self.stage() |
| 93 | self.assertEqual(sentinel.read_text(), "untouched") |
| 94 | |
| 95 | |
| 96 | if __name__ == "__main__": |
| 97 | unittest.main() |