from __future__ import annotations from dataclasses import replace from contextlib import redirect_stdout import hashlib import importlib.util import io import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest import mock PROJECT_ROOT = Path(__file__).parents[4] VALIDATOR_PATH = PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension_unpacked_validator.py" SPEC = importlib.util.spec_from_file_location("bili_authenticated_extension_unpacked_validator", VALIDATOR_PATH) assert SPEC is not None and SPEC.loader is not None validator = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = validator SPEC.loader.exec_module(validator) def _remove_reparse(path: Path) -> None: try: if path.is_symlink(): path.unlink() elif path.exists(): os.rmdir(path) except FileNotFoundError: pass class UnpackedProjectionTests(unittest.TestCase): maxDiff = None def setUp(self) -> None: temp_parent = PROJECT_ROOT / "dev" / "tmp" temp_parent.mkdir(parents=True, exist_ok=True) self.temp = tempfile.TemporaryDirectory(dir=temp_parent) self.addCleanup(self.temp.cleanup) self.temp_root = Path(self.temp.name) def _project(self, parent: str = "case") -> Path: root = self.temp_root / parent / validator.PROJECT_ID source = root.joinpath(*validator.SOURCE_ROOT_REL.parts) projection = root.joinpath(*validator.PROJECTION_ROOT_REL.parts) source.mkdir(parents=True) projection.mkdir(parents=True) (root / "mbx.project.yaml").write_text("schema_version: '1.0'\nproject:\n id: project-info\n", encoding="utf-8") authoritative = PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension" for name in validator.EXPECTED_FILES: shutil.copyfile(authoritative / name, source / name) shutil.copyfile(authoritative / name, projection / name) shutil.copyfile( authoritative / "source-artifact-manifest.json", root.joinpath(*validator.SOURCE_ARTIFACT_MANIFEST_REL.parts), ) shutil.copyfile( PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension_unpacked_contract.json", root.joinpath(*validator.CONTRACT_REL.parts), ) return root def _assert_error(self, root: Path, expected: str, **kwargs: object) -> None: with self.assertRaises(validator.ValidationError) as caught: validator.validate_project(str(root), **kwargs) self.assertEqual(caught.exception.code, expected) def _assert_cli_safety_stop(self, root: Path, expected: str) -> None: output = io.StringIO() with redirect_stdout(output): code = validator.main(["--project-root", str(root)]) body = output.getvalue() self.assertEqual(code, 3) self.assertNotIn("VALIDATION_PASS_ONLY", body) self.assertEqual(json.loads(body), {"schema": 1, "status": "SAFETY_STOP", "error_code": expected}) def _case_only_rename(self, path: Path, new_name: str) -> Path: temporary = path.with_name(f"case-rename-{path.name}.tmp") renamed = path.with_name(new_name) os.replace(path, temporary) os.replace(temporary, renamed) self.assertFalse(path.exists() and path.name == new_name) actual_names = tuple(entry.name for entry in os.scandir(renamed.parent)) self.assertIn(new_name, actual_names) return renamed def _junction(self, link: Path, target: Path) -> None: result = subprocess.run( ["cmd.exe", "/d", "/c", "mklink", "/J", str(link), str(target)], capture_output=True, text=True, check=False, ) if result.returncode != 0: self.fail(f"real junction creation failed: {result.returncode}") self.addCleanup(_remove_reparse, link) attributes = os.lstat(link).st_file_attributes self.assertTrue(attributes & validator.FILE_ATTRIBUTE_REPARSE_POINT) def _leaf_reparse(self, link: Path, target: Path) -> None: result = subprocess.run( ["cmd.exe", "/d", "/c", "mklink", str(link), str(target)], capture_output=True, text=True, check=False, ) if result.returncode != 0: junction_target = target.with_name(f"{target.name}-junction-target") junction_target.mkdir() self._junction(link, junction_target) else: self.addCleanup(_remove_reparse, link) attributes = os.lstat(link).st_file_attributes self.assertTrue(attributes & validator.FILE_ATTRIBUTE_REPARSE_POINT) def test_01_real_projection_passes_exact_set_bytes_tree_and_id(self) -> None: result = validator.validate_project(str(PROJECT_ROOT)) self.assertEqual(result["status"], "VALIDATION_PASS_ONLY") self.assertEqual(result["extension_id"], "oidmclckpdmpabbfedplkbdplmfcenbb") self.assertEqual(result["file_count"], 5) self.assertEqual( result["tree_sha256"], "0F5095CFC0CD62E0165E0A990104D9E46DA5F04B42C89B16D3F7A3420ADDB968", ) projection = PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension_unpacked" self.assertEqual(tuple(sorted(path.name for path in projection.iterdir())), validator.EXPECTED_FILES) self.assertFalse(any(part.startswith("_") for path in projection.iterdir() for part in path.parts[-1:])) self.assertEqual( result["source_artifact_manifest"], { "bytes": 3770, "sha256": "132B637634550A43B0EDD5E8E74C439205DE4BFECCE0A9276D8EBEF78445ECFC", }, ) def test_02_reserved_extra_missing_and_byte_drift_fail_closed(self) -> None: root = self._project("reserved") projection = root.joinpath(*validator.PROJECTION_ROOT_REL.parts) (projection / "__reserved").write_bytes(b"x") self._assert_error(root, "E_RESERVED_NAME") root = self._project("missing") root.joinpath(*validator.PROJECTION_ROOT_REL.parts, "sidepanel.css").unlink() self._assert_error(root, "E_PATH_MISSING") root = self._project("drift") with root.joinpath(*validator.PROJECTION_ROOT_REL.parts, "background.js").open("ab") as handle: handle.write(b"\n") self._assert_error(root, "E_BYTE_DRIFT") def test_03_project_root_reparse_fails_before_content_acceptance(self) -> None: target = self._project("target") alias_parent = self.temp_root / "alias" alias_parent.mkdir() alias = alias_parent / validator.PROJECT_ID self._junction(alias, target) self._assert_error(alias, "E_REPARSE") def test_04_source_root_reparse_fails_before_source_read(self) -> None: root = self._project("source-root") source = root.joinpath(*validator.SOURCE_ROOT_REL.parts) external = self.temp_root / "source-external" shutil.copytree(source, external) shutil.rmtree(source) self._junction(source, external) with mock.patch.object(Path, "read_bytes", side_effect=AssertionError("content read before gate")): self._assert_error(root, "E_REPARSE") def test_05_projection_root_reparse_fails_before_handoff(self) -> None: root = self._project("projection-root") projection = root.joinpath(*validator.PROJECTION_ROOT_REL.parts) external = self.temp_root / "projection-external" shutil.copytree(projection, external) shutil.rmtree(projection) self._junction(projection, external) self._assert_error(root, "E_REPARSE") def test_06_intermediate_parent_junction_fails(self) -> None: root = self._project("parent") dev = root / "dev" external = self.temp_root / "dev-external" shutil.move(str(dev), str(external)) self._junction(dev, external) self._assert_error(root, "E_REPARSE") def test_07_source_and_projection_leaf_real_reparse_points_fail(self) -> None: for parent_rel, case_name in ( (validator.SOURCE_ROOT_REL, "source-leaf"), (validator.PROJECTION_ROOT_REL, "projection-leaf"), ): with self.subTest(parent=parent_rel.as_posix()): root = self._project(case_name) leaf = root.joinpath(*parent_rel.parts, "manifest.json") external = self.temp_root / f"{case_name}-manifest.json" shutil.copyfile(leaf, external) leaf.unlink() self._leaf_reparse(leaf, external) self._assert_error(root, "E_REPARSE") def test_08_unknown_reparse_tag_adapter_fails(self) -> None: root = self._project("unknown-tag") target = root.joinpath(*validator.PROJECTION_ROOT_REL.parts, "background.js") target_inode = os.lstat(target).st_ino original = validator._reparse_tag def probe(st: os.stat_result) -> int: if st.st_ino == target_inode: return 0xDEADBEEF return original(st) with mock.patch.object(validator, "_reparse_tag", side_effect=probe): self._assert_error(root, "E_REPARSE") def test_09_physical_escape_adapter_fails_component_boundary(self) -> None: root = self._project("physical-escape") original = validator._open_identity def open_identity(*args: object, **kwargs: object): identity = original(*args, **kwargs) if identity.relative_path == validator.PROJECTION_ROOT_REL.as_posix(): return replace(identity, final_path=str(root.parent / "project-info-escape")) return identity with mock.patch.object(validator, "_open_identity", side_effect=open_identity): self._assert_error(root, "E_PATH_ESCAPE") def test_10_same_byte_leaf_replacement_between_passes_fails(self) -> None: root = self._project("identity-race") leaf = root.joinpath(*validator.PROJECTION_ROOT_REL.parts, "sidepanel.js") def replace_same_bytes() -> None: replacement = leaf.with_suffix(".replacement") replacement.write_bytes(leaf.read_bytes()) os.replace(replacement, leaf) self._assert_error(root, "E_PATH_IDENTITY_DRIFT", between_passes=replace_same_bytes) def test_11_manifest_key_drift_cannot_derive_accepted_identity(self) -> None: manifest_path = PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension_unpacked" / "manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["key"] = manifest["key"][:-1] + ("A" if manifest["key"][-1] != "A" else "B") with self.assertRaises(validator.ValidationError) as caught: validator._validate_manifest(json.dumps(manifest).encode("utf-8")) self.assertEqual(caught.exception.code, "E_MANIFEST") def test_12_source_parent_root_and_all_five_leaf_case_drift_fail_closed(self) -> None: cases = ( (Path("dev"), "Dev"), (Path(*validator.SOURCE_ROOT_REL.parts), "Bili_authenticated_extension"), *( (Path(*validator.SOURCE_ROOT_REL.parts, name), name[0].upper() + name[1:]) for name in validator.EXPECTED_FILES ), ) for index, (relative, new_name) in enumerate(cases): with self.subTest(relative=relative.as_posix()): root = self._project(f"source-case-{index}") target = root / relative self._case_only_rename(target, new_name) self._assert_cli_safety_stop(root, "E_PATH_CASE_DRIFT") def test_13_projection_case_only_rename_fails_closed(self) -> None: root = self._project("projection-case") target = root.joinpath(*validator.PROJECTION_ROOT_REL.parts, "manifest.json") self._case_only_rename(target, "Manifest.json") self._assert_cli_safety_stop(root, "E_PATH_CASE_DRIFT") def test_14_source_artifact_manifest_exactly_binds_five_payload_entries(self) -> None: path = PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension" / "source-artifact-manifest.json" data = path.read_bytes() self.assertEqual(len(data), validator.SOURCE_ARTIFACT_MANIFEST_BYTES) self.assertEqual(hashlib.sha256(data).hexdigest().upper(), validator.SOURCE_ARTIFACT_MANIFEST_SHA256) source_manifest = json.loads(data) contract = json.loads( (PROJECT_ROOT / "dev" / "project-dev" / "bili_authenticated_extension_unpacked_contract.json").read_bytes() ) expected = {entry["path"]: entry for entry in contract["files"]} observed = {entry["path"]: entry for entry in source_manifest["files"] if entry["path"] in expected} self.assertEqual(tuple(sorted(observed)), validator.EXPECTED_FILES) self.assertEqual(observed, expected) if __name__ == "__main__": unittest.main()