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()
|
