from __future__ import annotations import base64 import hashlib import json import os from pathlib import Path import shutil import signal import subprocess import tempfile import time import unittest import zipfile PROJECT = Path(__file__).resolve().parents[4] SOURCE = PROJECT / "dev" / "project-dev" / "bili_authenticated_extension_managed_load" BUILDER = SOURCE / "build_webstore_upload.ps1" INSTALLER = SOURCE / "install_managed_extension_policy.ps1" CONTRACT_PATH = SOURCE / "managed-load-contract.json" POWERSHELL = shutil.which("powershell.exe") or shutil.which("pwsh.exe") SENTINEL = "ZXQ_SECRET_SENTINEL_20260810" def sha_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest().upper() def sha_file(path: Path) -> str: return sha_bytes(path.read_bytes()) def dump_json(path: Path, value: object) -> tuple[int, str]: data = (json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n").encode() path.write_bytes(data) return len(data), sha_bytes(data) def tree_summary(root: Path) -> tuple[int, str]: entries = [] for path in root.rglob("*"): if path.is_file(): rel = path.relative_to(root).as_posix() entries.append((rel, path.stat().st_size, sha_file(path))) serial = b"".join( rel.encode() + b"\0" + str(size).encode() + b"\0" + digest.lower().encode() + b"\n" for rel, size, digest in sorted(entries) ) return len(entries), sha_bytes(serial) def canonical_audit_section(path: Path, audit_id: str, contract: dict) -> bytes: raw = path.read_bytes() if raw.startswith(b"\xef\xbb\xbf"): raise ValueError("audit BOM") text = raw.decode("utf-8", errors="strict") if "\0" in text or "\r" in text: raise ValueError("audit must be LF-only") section_contract = contract["trust"]["implementation_audit_section_contract"] prefix = section_contract["heading_prefix"] heading = prefix + audit_id lines = text.split("\n") starts = [index for index, line in enumerate(lines) if line == heading] if len(starts) != 1: raise ValueError("audit ID not unique") start = starts[0] end = next((index for index in range(start + 1, len(lines)) if lines[index].startswith(prefix)), len(lines)) while end > start and lines[end - 1] == "": end -= 1 return ("\n".join(lines[start:end]) + "\n").encode("utf-8") def payload_summary(contract: dict) -> tuple[list[dict], str]: entries = [{k: contract["overlay_manifest"][k] for k in ("path", "bytes", "sha256")}] entries += [{k: item[k] for k in ("path", "bytes", "sha256")} for item in contract["runtime_payload"]] entries += [{k: item[k] for k in ("path", "bytes", "sha256")} for item in contract["icons"]] entries.sort(key=lambda item: item["path"]) serial = b"".join( item["path"].encode() + b"\0" + str(item["bytes"]).encode() + b"\0" + item["sha256"].lower().encode() + b"\n" for item in entries ) return entries, sha_bytes(serial) def canonical_preimage(key_exists: bool, target: str, values: list[dict]) -> dict: summaries = [] for value in sorted(values, key=lambda item: item["name"]): raw = base64.b64decode(value["data_base64"]) summaries.append({ "name": value["name"], "kind": value["kind"], "data_bytes": len(raw), "data_sha256": sha_bytes(raw), }) serial = ("true" if key_exists else "false").encode() + b"\n" serial += b"".join( item["name"].encode() + b"\0" + item["kind"].encode() + b"\0" + str(item["data_bytes"]).encode() + b"\0" + item["data_sha256"].lower().encode() + b"\n" for item in summaries ) return { "key_exists": key_exists, "target_value_name": target, "target_absent": True, "values": summaries, "canonical_sha256": sha_bytes(serial), } def instrument_test_copy(source: Path, destination: Path, kind: str) -> None: text = source.read_text(encoding="utf-8") if kind == "builder": text = text.replace( " [switch]$Build\n # INTERNAL_TEST_ADAPTER_PARAMETER_ANCHOR", " [switch]$Build,\n" " [Parameter(DontShow = $true)][switch]$AdapterAllowTestScope,\n" " [Parameter(DontShow = $true)][string]$AdapterOriginalSourceRoot", ) context = ( "$script:InternalAllowTestScope = [bool]$AdapterAllowTestScope\n" "$script:InternalOriginalSourceRoot = $AdapterOriginalSourceRoot" ) else: text = text.replace( " [switch]$Rollback\n # INTERNAL_TEST_ADAPTER_PARAMETER_ANCHOR", " [switch]$Rollback,\n" " [Parameter(DontShow = $true)][switch]$AdapterAllowTestScope,\n" " [Parameter(DontShow = $true)][string]$AdapterFileRegistryPath,\n" " [Parameter(DontShow = $true)][string]$AdapterInjectFailure = 'none',\n" " [Parameter(DontShow = $true)][string]$AdapterPauseAt = 'none',\n" " [Parameter(DontShow = $true)][string]$AdapterPauseMarker,\n" " [Parameter(DontShow = $true)][switch]$AdapterFailureEvidenceWriteFailure", ) context = ( "$script:InternalAllowTestScope = [bool]$AdapterAllowTestScope\n" "$script:InternalFileRegistryPath = $AdapterFileRegistryPath\n" "$script:InternalInjectFailure = $AdapterInjectFailure\n" "$script:InternalPauseAt = $AdapterPauseAt\n" "$script:InternalPauseMarker = $AdapterPauseMarker\n" "$script:InternalFailureEvidenceWriteFailure = [bool]$AdapterFailureEvidenceWriteFailure" ) anchor = "# INTERNAL_TEST_ADAPTER_CONTEXT_ANCHOR" if anchor not in text or "INTERNAL_TEST_ADAPTER_PARAMETER_ANCHOR" in text: raise AssertionError("test adapter anchors did not match") destination.write_text(text.replace(anchor, context), encoding="utf-8", newline="\n") @unittest.skipUnless(POWERSHELL, "PowerShell is required") class ManagedLoadTests(unittest.TestCase): maxDiff = None def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self._prepare_tree() def _prepare_tree(self) -> None: self.root = Path(self.temp.name) self.project = self.root / "project" self.managed = self.project / "dev" / "project-dev" / "bili_authenticated_extension_managed_load" self.original = self.project / "dev" / "project-dev" / "bili_authenticated_extension" self.managed.parent.mkdir(parents=True) shutil.copytree(SOURCE, self.managed) shutil.copytree(PROJECT / "dev" / "project-dev" / "bili_authenticated_extension", self.original) self.builder = self.managed / "build_webstore_upload.ps1" self.installer = self.managed / "install_managed_extension_policy.ps1" instrument_test_copy(BUILDER, self.builder, "builder") instrument_test_copy(INSTALLER, self.installer, "installer") self.contract_path = self.managed / "managed-load-contract.json" self.contract = json.loads(self.contract_path.read_text(encoding="utf-8")) self.audit = self.project / "dev-doc" / "开发审计报告.md" self.audit.parent.mkdir(parents=True) self.source_approval = self.project / self.contract["trust"]["reviewer_source_approval_path"] self.source_approval.parent.mkdir(parents=True) self.historical_source_approval = self.project / self.contract["trust"]["historical_reviewer_source_approval_path"] self._write_source_approval(self.source_approval) def tearDown(self) -> None: self.temp.cleanup() def _reset(self) -> None: self.temp.cleanup() self.temp = tempfile.TemporaryDirectory() self._prepare_tree() def _write_source_approval(self, path: Path, extra: dict | None = None) -> tuple[int, str]: path.parent.mkdir(parents=True, exist_ok=True) count, tree_hash = tree_summary(self.managed) entries, payload_hash = payload_summary(self.contract) original_manifest = self.original / "source-artifact-manifest.json" audit_id = "DEV-AUDIT-TEST-MANAGED-LOAD-IMPLEMENTATION-20260810-001" self.audit.write_text( f"## {audit_id}\n\n" f"- task={self.contract['task_id']}\n" "- verdict=PASS/0\n" "- blocking_findings=0\n" f"- managed_load_source_tree={tree_hash}\n", encoding="utf-8", newline="\n", ) value = { "schema": self.contract["schemas"]["source_approval_schema"], "scope": "test-only-controlled-webstore-upload-source", "status": "APPROVED", "task_id": self.contract["task_id"], "approved_by_role": "dev.reviewer.project", "managed_load_contract": {"path": self.contract["trust"]["managed_load_contract_path"], "bytes": self.contract_path.stat().st_size, "sha256": sha_file(self.contract_path)}, "original_source_manifest": {"path": self.contract["trust"]["original_source_manifest_path"], "bytes": original_manifest.stat().st_size, "sha256": sha_file(original_manifest)}, "managed_load_source_tree": {"file_count": count, "tree_sha256": tree_hash}, "webstore_payload": {"entry_count": 9, "payload_tree_sha256": payload_hash, "entries": entries}, "implementation_review": { "audit_id": audit_id, "audit_path": self.contract["trust"]["implementation_audit_path"], "audit_section_format": self.contract["trust"]["implementation_audit_section_contract"]["format"], "audit_section_bytes": len(canonical_audit_section(self.audit, audit_id, self.contract)), "audit_section_sha256": sha_bytes(canonical_audit_section(self.audit, audit_id, self.contract)), "verdict": "PASS", }, "approved_at": "2026-08-10T00:00:00.0000000+00:00", } if extra: value.update(extra) return dump_json(path, value) def _run(self, script: Path, *args: str, timeout: float = 30) -> subprocess.CompletedProcess[str]: return subprocess.run( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(script), *map(str, args)], text=True, capture_output=True, timeout=timeout, cwd=PROJECT, ) def _source_args(self) -> list[str]: return [ "-ApprovedSourceReceipt", str(self.source_approval), "-ApprovedSourceReceiptBytes", str(self.source_approval.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(self.source_approval), "-AdapterAllowTestScope", ] def _make_install_fixture(self, key_exists: bool = True) -> dict: build_root = self.root / "build" result = self._run(self.builder, *self._source_args(), "-Build", "-OutputRoot", str(build_root)) self.assertEqual(result.returncode, 0, (result.stdout, result.stderr)) receipt = build_root / "webstore-upload-build-receipt.json" receipt_json = json.loads(receipt.read_text(encoding="utf-8")) evidence = self.root / "release-evidence.json" dump_json(evidence, { "schema": 1, "item_id": self.contract["extension_id"], "version": self.contract["extension_version"], "publication_state": "PUBLISHED", "store_url": f"https://chromewebstore.google.com/detail/{self.contract['extension_id']}", "published_at": "2026-08-10T00:00:00.0000000+00:00", "upload_zip_sha256": receipt_json["zip"]["sha256"], }) release = self.root / "release-approval.json" dump_json(release, { "schema": 1, "scope": "test-only-install-exact-webstore-extension", "status": "APPROVED", "task_id": self.contract["task_id"], "approved_by_role": "dev.reviewer.project", "extension_id": self.contract["extension_id"], "extension_version": self.contract["extension_version"], "webstore_update_url": self.contract["webstore_update_url"], "managed_load_contract": {"bytes": self.contract_path.stat().st_size, "sha256": sha_file(self.contract_path)}, "upload_build_receipt": {"path": str(receipt), "bytes": receipt.stat().st_size, "sha256": sha_file(receipt)}, "release_evidence": {"path": str(evidence), "bytes": evidence.stat().st_size, "sha256": sha_file(evidence)}, "approved_at": "2026-08-10T00:00:00.0000000+00:00", }) receipt_root = self.root / "receipts" receipt_root.mkdir() failure_root = self.root / "failures" failure_root.mkdir() values = [] if key_exists: values = [ {"name": "3", "kind": "Binary", "data_base64": base64.b64encode(b"\x00\x01\xff").decode()}, {"name": "2", "kind": "String", "data_base64": base64.b64encode("other\0".encode("utf-16-le")).decode()}, ] registry = self.root / "registry.json" dump_json(registry, {"key_exists": key_exists, "values": values}) preimage = canonical_preimage(key_exists, "7", values) raw_policy = (self.contract["policy_data"] + "\0").encode("utf-16-le") paths = { "parent": str(receipt_root), "failure_root": str(failure_root), "install_pending": str(receipt_root / "managed-extension-install-pending.json"), "install_final": str(receipt_root / "managed-extension-install-receipt.json"), "install_recovery": str(receipt_root / "managed-extension-install-recovery-receipt.json"), "rollback_pending": str(receipt_root / "managed-extension-rollback-pending.json"), "rollback_final": str(receipt_root / "managed-extension-rollback-receipt.json"), "rollback_recovery": str(receipt_root / "managed-extension-rollback-recovery-receipt.json"), } approval = self.root / "install-approval.json" dump_json(approval, { "schema": 1, "scope": "test-only-install-exact-managed-extension-policy", "status": "APPROVED", "task_id": self.contract["task_id"], "approved_by_role": "project.admin", "release_approval": {"bytes": release.stat().st_size, "sha256": sha_file(release)}, "source_approval": {"bytes": self.source_approval.stat().st_size, "sha256": sha_file(self.source_approval)}, "policy": {"key": self.contract["policy_key"], "value_name": "7", "kind": "String", "data_bytes": len(raw_policy), "data_sha256": sha_bytes(raw_policy)}, "policy_preimage": preimage, "native_host": self.contract["native_host"], "receipt_paths": paths, "approved_at": "2026-08-10T00:00:00.0000000+00:00", }) return {"release": release, "approval": approval, "registry": registry, "paths": paths} def _install_args(self, fixture: dict) -> list[str]: return [ "-ApprovedSourceReceipt", str(self.source_approval), "-ApprovedSourceReceiptBytes", str(self.source_approval.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(self.source_approval), "-ReleaseApproval", str(fixture["release"]), "-ReleaseApprovalBytes", str(fixture["release"].stat().st_size), "-ReleaseApprovalSha256", sha_file(fixture["release"]), "-InstallApproval", str(fixture["approval"]), "-InstallApprovalBytes", str(fixture["approval"].stat().st_size), "-InstallApprovalSha256", sha_file(fixture["approval"]), "-AdapterAllowTestScope", "-AdapterFileRegistryPath", str(fixture["registry"]), ] def _wait_for_marker(self, marker: Path, timeout: float = 20) -> None: deadline = time.monotonic() + timeout while not marker.exists() and time.monotonic() < deadline: time.sleep(0.05) self.assertTrue(marker.exists(), f"pause marker not reached: {marker.name}") def _start_paused(self, args: list[str], mode: str, point: str, marker: Path) -> subprocess.Popen[str]: return subprocess.Popen( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(self.installer), *args, mode, "-AdapterPauseAt", point, "-AdapterPauseMarker", str(marker)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=PROJECT, creationflags=getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0), ) def _kill_paused(self, args: list[str], mode: str, point: str, *, ctrl_break: bool = False) -> tuple[str, str]: marker = self.root / (point + ".marker") process = self._start_paused(args, mode, point, marker) self._wait_for_marker(marker) if ctrl_break: process.send_signal(signal.CTRL_BREAK_EVENT) else: process.kill() try: return process.communicate(timeout=5 if ctrl_break else 15) except subprocess.TimeoutExpired: process.kill() process.communicate(timeout=10) if ctrl_break: raise unittest.SkipTest("Windows PowerShell host did not deliver CTRL_BREAK_EVENT to the paused pipeline") raise def _failure_records(self, fixture: dict) -> list[dict]: root = Path(fixture["paths"]["failure_root"]) return [json.loads(path.read_text(encoding="utf-8")) for path in sorted(root.glob("managed-extension-failure-*.json"))] def _assert_failure_record(self, fixture: dict, record: dict, primary: str, secondary: str | None) -> None: self.assertEqual(set(record), set(self.contract["schemas"]["failure_evidence_keys"])) self.assertEqual(record["status"], "FAILED") self.assertEqual(record["task_id"], self.contract["task_id"]) self.assertEqual(record["primary_error_code"], primary) self.assertEqual(record["primary_error_sha256"], sha_bytes(primary.encode())) self.assertEqual(record["secondary_error_code"], secondary) self.assertEqual(record["secondary_error_sha256"], None if secondary is None else sha_bytes(secondary.encode())) serialized = json.dumps(record, ensure_ascii=False) for forbidden in (str(self.root), str(fixture["registry"]), SENTINEL, "Injected terminating error"): self.assertNotIn(forbidden, serialized) def test_01_builder_exact_nine_and_secret_fail_closed(self) -> None: valid = self._run(self.builder, *self._source_args()) self.assertEqual(valid.returncode, 0, (valid.stdout, valid.stderr)) self.assertEqual(json.loads(valid.stdout)["status"], "VALIDATION_PASS_ONLY") output = self.root / "upload" built = self._run(self.builder, *self._source_args(), "-Build", "-OutputRoot", str(output)) self.assertEqual(built.returncode, 0, (built.stdout, built.stderr)) self.assertEqual(sorted(item.name for item in output.iterdir()), ["project-info-bili-auth-ingress-webstore.zip", "webstore-upload-build-receipt.json"]) with zipfile.ZipFile(output / "project-info-bili-auth-ingress-webstore.zip") as archive: self.assertEqual(sorted(archive.namelist()), sorted(item["path"] for item in payload_summary(self.contract)[0])) self.assertEqual(len(archive.infolist()), 9) bad = self.root / "bad-approval.json" self._write_source_approval(bad, {"unexpected": SENTINEL}) failed = self._run( self.builder, "-ApprovedSourceReceipt", str(bad), "-ApprovedSourceReceiptBytes", str(bad.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(bad), "-AdapterAllowTestScope", ) self.assertEqual(failed.returncode, 3) self.assertNotIn(SENTINEL, failed.stdout + failed.stderr) duplicate = self.root / "duplicate-approval.json" raw = self.source_approval.read_text(encoding="utf-8").replace( '"status":"APPROVED"', '"status":"APPROVED","statu\\u0073":"APPROVED"', 1, ).encode() duplicate.write_bytes(raw) failed = self._run( self.builder, "-ApprovedSourceReceipt", str(duplicate), "-ApprovedSourceReceiptBytes", str(len(raw)), "-ApprovedSourceReceiptSha256", sha_bytes(raw), "-AdapterAllowTestScope", ) self.assertEqual(failed.returncode, 3) def test_02_install_health_rollback_preserves_full_preimage(self) -> None: fixture = self._make_install_fixture(key_exists=True) args = self._install_args(fixture) before = fixture["registry"].read_bytes() validate = self._run(self.installer, *args) self.assertEqual(validate.returncode, 0, (validate.stdout, validate.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) install = self._run(self.installer, *args, "-Install") self.assertEqual(install.returncode, 0, (install.stdout, install.stderr)) self.assertEqual(json.loads(install.stdout)["status"], "INSTALL_COMPLETE") health = self._run(self.installer, *args, "-HealthCheck") self.assertEqual(health.returncode, 0, (health.stdout, health.stderr)) rollback = self._run(self.installer, *args, "-Rollback") self.assertEqual(rollback.returncode, 0, (rollback.stdout, rollback.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["install_pending"]).exists()) self.assertTrue(Path(fixture["paths"]["install_final"]).exists()) self.assertTrue(Path(fixture["paths"]["rollback_pending"]).exists()) self.assertTrue(Path(fixture["paths"]["rollback_final"]).exists()) def test_03_ordinary_failure_and_crash_reopen_recover(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() failed = self._run(self.installer, *args, "-Install", "-AdapterInjectFailure", "after-policy-write") self.assertEqual(failed.returncode, 4, (failed.stdout, failed.stderr)) self.assertEqual(json.loads(failed.stdout)["status"], "FAILED_ROLLED_BACK") self.assertEqual(before, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) records = self._failure_records(fixture) self.assertEqual(len(records), 1) self._assert_failure_record(fixture, records[0], "E_MANAGED_LOAD", None) # A separate fixture proves real process termination is recovered only on reopen. self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() marker = self.root / "pause.marker" process = subprocess.Popen( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(self.installer), *args, "-Install", "-AdapterPauseAt", "after-policy-write", "-AdapterPauseMarker", str(marker)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=PROJECT, ) deadline = time.monotonic() + 20 while not marker.exists() and time.monotonic() < deadline: time.sleep(0.05) self.assertTrue(marker.exists(), "pause marker not reached") process.kill() process.communicate(timeout=10) reopened = self._run(self.installer, *args) self.assertEqual(reopened.returncode, 4, (reopened.stdout, reopened.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["install_recovery"]).exists()) def test_04_policy_race_and_release_secret_are_zero_side_effect(self) -> None: fixture = self._make_install_fixture(key_exists=True) args = self._install_args(fixture) before = fixture["registry"].read_bytes() state = json.loads(before) state["values"].append({"name": "7", "kind": "String", "data_base64": base64.b64encode("race\0".encode("utf-16-le")).decode()}) dump_json(fixture["registry"], state) raced = fixture["registry"].read_bytes() failed = self._run(self.installer, *args, "-Install") self.assertEqual(failed.returncode, 3) self.assertEqual(raced, fixture["registry"].read_bytes()) self.assertFalse(any(Path(fixture["paths"][name]).exists() for name in ("install_pending", "install_final"))) evidence_binding = json.loads(fixture["release"].read_text(encoding="utf-8"))["release_evidence"] evidence = Path(evidence_binding["path"]) injected = json.loads(evidence.read_text(encoding="utf-8")) injected["unexpected"] = SENTINEL dump_json(evidence, injected) release = json.loads(fixture["release"].read_text(encoding="utf-8")) release["release_evidence"] = {"path": str(evidence), "bytes": evidence.stat().st_size, "sha256": sha_file(evidence)} dump_json(fixture["release"], release) # The install approval is now stale, so both the secret schema and exact binding fail before writes. failed = self._run(self.installer, *self._install_args(fixture)) self.assertEqual(failed.returncode, 3) self.assertNotIn(SENTINEL, failed.stdout + failed.stderr) def test_05_managed_source_extra_and_reparse_fail_after_approval(self) -> None: nested = self.managed / "nested" nested.mkdir() (nested / "extra.ps1").write_text("throw 'not approved'\n", encoding="utf-8") failed = self._run(self.builder, *self._source_args()) self.assertEqual(failed.returncode, 3) self.assertEqual(list(self.root.glob("*.zip")), []) shutil.rmtree(nested) link = self.managed / "reparse-link" try: os.symlink(self.original, link, target_is_directory=True) except OSError: created = subprocess.run( ["cmd.exe", "/d", "/c", "mklink", "/J", str(link), str(self.original)], capture_output=True, text=True, ) self.assertEqual(created.returncode, 0, (created.stdout, created.stderr)) failed = self._run(self.builder, *self._source_args()) self.assertEqual(failed.returncode, 3) def test_06_all_registry_kinds_round_trip_exactly(self) -> None: fixture = self._make_install_fixture(key_exists=True) values = [ {"name": "1", "kind": "String", "data_base64": base64.b64encode("plain\0".encode("utf-16-le")).decode()}, {"name": "2", "kind": "ExpandString", "data_base64": base64.b64encode("%LOCALAPPDATA%\0".encode("utf-16-le")).decode()}, {"name": "3", "kind": "MultiString", "data_base64": base64.b64encode("a\0b\0\0".encode("utf-16-le")).decode()}, {"name": "4", "kind": "DWord", "data_base64": base64.b64encode((7).to_bytes(4, "little", signed=True)).decode()}, {"name": "5", "kind": "QWord", "data_base64": base64.b64encode((9).to_bytes(8, "little", signed=True)).decode()}, {"name": "6", "kind": "Binary", "data_base64": base64.b64encode(b"\x00\x80\xff").decode()}, ] before = dump_json(fixture["registry"], {"key_exists": True, "values": values}) approval = json.loads(fixture["approval"].read_text(encoding="utf-8")) approval["policy_preimage"] = canonical_preimage(True, "7", values) dump_json(fixture["approval"], approval) before_bytes = fixture["registry"].read_bytes() args = self._install_args(fixture) self.assertEqual(self._run(self.installer, *args, "-Install").returncode, 0) self.assertEqual(self._run(self.installer, *args, "-HealthCheck").returncode, 0) self.assertEqual(self._run(self.installer, *args, "-Rollback").returncode, 0) self.assertEqual(before_bytes, fixture["registry"].read_bytes()) def test_07_unknown_temp_and_tampered_pending_are_ambiguous(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() unknown = Path(fixture["paths"]["install_pending"] + "." + "a" * 32 + ".tmp") unknown.write_bytes(b"{}\n") failed = self._run(self.installer, *args) self.assertEqual(failed.returncode, 3) self.assertEqual(json.loads(failed.stdout)["error_code"], "E_RECEIPT") self.assertEqual(before, fixture["registry"].read_bytes()) records = self._failure_records(fixture) self.assertEqual(len(records), 1) self._assert_failure_record(fixture, records[0], "E_RECEIPT", None) unknown.unlink() installed = self._run(self.installer, *args, "-Install") self.assertEqual(installed.returncode, 0) final = Path(fixture["paths"]["install_final"]) final.unlink() pending = Path(fixture["paths"]["install_pending"]) corrupted = json.loads(pending.read_text(encoding="utf-8")) corrupted["source_approval"]["sha256"] = "0" * 64 dump_json(pending, corrupted) active = fixture["registry"].read_bytes() failed = self._run(self.installer, *args) self.assertEqual(failed.returncode, 3) self.assertEqual(json.loads(failed.stdout)["error_code"], "E_RECEIPT") self.assertEqual(active, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_recovery"]).exists()) records = self._failure_records(fixture) self.assertEqual(len(records), 2) self._assert_failure_record(fixture, records[-1], "E_RECEIPT", None) def test_08_rollback_crash_reopen_restores_active_state(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) self.assertEqual(self._run(self.installer, *args, "-Install").returncode, 0) active = fixture["registry"].read_bytes() marker = self.root / "rollback-pause.marker" process = subprocess.Popen( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(self.installer), *args, "-Rollback", "-AdapterPauseAt", "after-policy-delete", "-AdapterPauseMarker", str(marker)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=PROJECT, ) deadline = time.monotonic() + 20 while not marker.exists() and time.monotonic() < deadline: time.sleep(0.05) self.assertTrue(marker.exists(), "rollback pause marker not reached") process.kill() process.communicate(timeout=10) recovered = self._run(self.installer, *args, "-Rollback") self.assertEqual(recovered.returncode, 4, (recovered.stdout, recovered.stderr)) self.assertEqual(active, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["rollback_recovery"]).exists()) self.assertFalse(Path(fixture["paths"]["rollback_final"]).exists()) def test_09_final_receipt_rename_is_the_commit_point(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) result = self._run(self.installer, *args, "-Install", "-AdapterInjectFailure", "after-install-final") self.assertEqual(result.returncode, 0, (result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["status"], "INSTALL_COMPLETE") result = self._run(self.installer, *args, "-Rollback", "-AdapterInjectFailure", "after-rollback-final") self.assertEqual(result.returncode, 0, (result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["status"], "ROLLBACK_COMPLETE_POLICY_REMOVED") def test_10_static_scope_and_frozen_runtime_non_regression(self) -> None: product_text = "\n".join(path.read_text(encoding="utf-8") for path in (BUILDER, INSTALLER)) for forbidden in ("--load-extension", "chrome://extensions", "chrome://policy", "Invoke-WebRequest", "Start-Process chrome", "remote-debugging-port"): self.assertNotIn(forbidden.lower(), product_text.lower()) self.assertNotIn(SENTINEL, product_text) original_manifest = json.loads((PROJECT / "dev" / "project-dev" / "bili_authenticated_extension" / "source-artifact-manifest.json").read_text(encoding="utf-8")) original_root = PROJECT / "dev" / "project-dev" / "bili_authenticated_extension" actual = [path for path in original_root.rglob("*") if path.is_file() and path.name != "source-artifact-manifest.json"] self.assertEqual(len(actual), 17) for entry in original_manifest["files"]: path = original_root / entry["path"] self.assertEqual((path.stat().st_size, sha_file(path)), (entry["bytes"], entry["sha256"])) def test_11_mutex_prevents_concurrent_recovery_or_write(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() marker = self.root / "mutex-pause.marker" first = subprocess.Popen( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(self.installer), *args, "-Install", "-AdapterPauseAt", "after-install-pending", "-AdapterPauseMarker", str(marker)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=PROJECT, ) deadline = time.monotonic() + 20 while not marker.exists() and time.monotonic() < deadline: time.sleep(0.05) self.assertTrue(marker.exists(), "mutex pause marker not reached") second = self._run(self.installer, *args, timeout=20) self.assertEqual(second.returncode, 3, (second.stdout, second.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_recovery"]).exists()) first.kill() first.communicate(timeout=10) reopened = self._run(self.installer, *args) self.assertEqual(reopened.returncode, 4, (reopened.stdout, reopened.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) def test_12_install_ordinary_fault_points_and_pipeline_stop(self) -> None: cases = { "after-install-pending-temp": 3, "after-install-pending": 4, "before-policy-write": 4, "after-policy-write": 4, "after-install-postimage": 4, "after-install-final-temp": 4, "after-install-final": 0, } for point, expected_exit in cases.items(): with self.subTest(point=point): self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() result = self._run(self.installer, *args, "-Install", "-AdapterInjectFailure", point) self.assertEqual(result.returncode, expected_exit, (point, result.stdout, result.stderr)) if point == "after-install-final": self.assertTrue(Path(fixture["paths"]["install_final"]).exists()) self.assertNotEqual(before, fixture["registry"].read_bytes()) else: self.assertEqual(before, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() stopped = self._run(self.installer, *args, "-Install", "-AdapterInjectFailure", "pipeline-stop") self.assertNotIn("INSTALL_COMPLETE", stopped.stdout) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) reopened = self._run(self.installer, *args) self.assertEqual(reopened.returncode, 4, (reopened.stdout, reopened.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["install_recovery"]).exists()) def test_13_reviewer_trust_anchor_audit_and_nested_paths_fail_closed(self) -> None: variants = { "missing-contract-path": lambda value: value["managed_load_contract"].pop("path"), "wrong-contract-path": lambda value: value["managed_load_contract"].update(path="wrong/contract.json"), "missing-original-path": lambda value: value["original_source_manifest"].pop("path"), "wrong-original-path": lambda value: value["original_source_manifest"].update(path="wrong/manifest.json"), "fake-audit": lambda value: value["implementation_review"].update(audit_section_sha256="0" * 64), "old-audit": lambda value: value["implementation_review"].update(audit_id="DEV-AUDIT-TEST-MANAGED-LOAD-OLD-001"), "old-tree": lambda value: value["managed_load_source_tree"].update(tree_sha256="0" * 64), } approval_before = self.source_approval.read_bytes() for label, mutate in variants.items(): with self.subTest(label=label): value = json.loads(approval_before.decode("utf-8")) mutate(value) dump_json(self.source_approval, value) output = self.root / f"{label}-output" result = self._run( self.builder, *self._source_args(), "-Build", "-OutputRoot", str(output), ) self.assertEqual(result.returncode, 3, (label, result.stdout, result.stderr)) self.assertFalse(output.exists()) self.source_approval.write_bytes(approval_before) audit_before = self.audit.read_bytes() try: value = json.loads(self.source_approval.read_text(encoding="utf-8")) audit_id = value["implementation_review"]["audit_id"] self.audit.write_text( f"## {audit_id}\n\n" f"- task={self.contract['task_id']}\n" f"- narrative=the strings - verdict=PASS/0 and - blocking_findings=0 are not terminal fields\n" f"- managed_load_source_tree={value['managed_load_source_tree']['tree_sha256']}\n", encoding="utf-8", newline="\n", ) section = canonical_audit_section(self.audit, audit_id, self.contract) value["implementation_review"].update( audit_section_bytes=len(section), audit_section_sha256=sha_bytes(section), ) dump_json(self.source_approval, value) result = self._run( self.builder, *self._source_args(), ) self.assertEqual(result.returncode, 3, (result.stdout, result.stderr)) finally: self.audit.write_bytes(audit_before) self.source_approval.write_bytes(approval_before) for location in ("developer", "source", "output", "install"): with self.subTest(self_signed_location=location): self._reset() roots = { "developer": self.root / "developer", "source": self.managed, "output": self.root / "candidate-output", "install": self.root / "install-state", } roots[location].mkdir(parents=True, exist_ok=True) candidate = roots[location] / "self-signed-source-approval.json" candidate.write_bytes(self.source_approval.read_bytes()) output = self.root / "must-remain-absent" result = self._run( self.builder, "-ApprovedSourceReceipt", str(candidate), "-ApprovedSourceReceiptBytes", str(candidate.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(candidate), "-Build", "-OutputRoot", str(output), ) self.assertEqual(result.returncode, 3, (location, result.stdout, result.stderr)) self.assertFalse(output.exists()) def test_14_product_cli_rejects_every_test_seam_without_side_effect(self) -> None: builder_cases = [ ["-TestOnlyAllowSyntheticApproval"], ["-TestOriginalSourceRoot", str(self.original)], ["-AdapterAllowTestScope"], ["-AdapterOriginalSourceRoot", str(self.original)], ] builder_base = [ "-ApprovedSourceReceipt", str(self.source_approval), "-ApprovedSourceReceiptBytes", str(self.source_approval.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(self.source_approval), ] for case in builder_cases: with self.subTest(builder_parameter=case[0]): output = self.root / (case[0].lstrip("-") + "-output") result = self._run(BUILDER, *builder_base, *case, "-Build", "-OutputRoot", str(output)) self.assertNotEqual(result.returncode, 0) self.assertFalse(output.exists()) fixture = self._make_install_fixture(key_exists=False) before = fixture["registry"].read_bytes() installer_base = [item for item in self._install_args(fixture) if item not in ("-AdapterAllowTestScope", "-AdapterFileRegistryPath", str(fixture["registry"]))] installer_cases = [ ["-TestOnlyAllowSyntheticApprovals"], ["-TestFileRegistryPath", str(fixture["registry"])], ["-TestInjectFailure", "after-policy-write"], ["-TestPauseAt", "after-policy-write"], ["-TestPauseMarker", str(self.root / "marker")], ["-AdapterAllowTestScope"], ["-AdapterFileRegistryPath", str(fixture["registry"])], ["-AdapterInjectFailure", "after-policy-write"], ["-AdapterPauseAt", "after-policy-write"], ["-AdapterPauseMarker", str(self.root / "marker")], ["-AdapterFailureEvidenceWriteFailure"], ] for case in installer_cases: with self.subTest(installer_parameter=case[0]): result = self._run(INSTALLER, *installer_base, *case, "-Install") self.assertNotEqual(result.returncode, 0) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertEqual(list(Path(fixture["paths"]["parent"]).iterdir()), []) self.assertEqual(self._failure_records(fixture), []) product_text = BUILDER.read_text(encoding="utf-8") + INSTALLER.read_text(encoding="utf-8") for name in ("TestOnlyAllowSynthetic", "TestOriginalSourceRoot", "TestFileRegistryPath", "TestInjectFailure", "TestPauseAt", "TestPauseMarker", "AdapterAllowTestScope", "AdapterFileRegistryPath", "AdapterInjectFailure", "AdapterPauseAt", "AdapterPauseMarker"): self.assertNotIn(name, product_text) def test_15_recovery_taxonomy_secondary_failures_and_independent_evidence(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() result = self._run( self.installer, *args, "-Install", "-AdapterInjectFailure", "after-policy-write,before-recovery-receipt", ) self.assertEqual(result.returncode, 3, (result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_RECEIPT") self.assertEqual(before, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_recovery"]).exists()) records = self._failure_records(fixture) self.assertEqual(len(records), 1) self._assert_failure_record(fixture, records[0], "E_MANAGED_LOAD", "E_RECEIPT") self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) self._kill_paused(args, "-Install", "after-policy-write") ambiguous = json.loads(fixture["registry"].read_text(encoding="utf-8")) ambiguous["values"].append({ "name": "99", "kind": "Binary", "data_base64": base64.b64encode(b"external-drift").decode(), }) dump_json(fixture["registry"], ambiguous) ambiguous_bytes = fixture["registry"].read_bytes() result = self._run(self.installer, *args) self.assertEqual(result.returncode, 3, (result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_RECOVERY_AMBIGUOUS") self.assertEqual(ambiguous_bytes, fixture["registry"].read_bytes()) records = self._failure_records(fixture) self.assertEqual(len(records), 1) self._assert_failure_record(fixture, records[0], "E_RECOVERY_AMBIGUOUS", None) self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() result = self._run( self.installer, *args, "-Install", "-AdapterInjectFailure", "after-install-pending", "-AdapterFailureEvidenceWriteFailure", ) self.assertEqual(result.returncode, 3, (result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_RECOVERY_FAILED") self.assertEqual(before, fixture["registry"].read_bytes()) self.assertEqual(self._failure_records(fixture), []) def test_16_real_kill_reopen_covers_every_install_and_rollback_durable_point(self) -> None: install_points = ( "after-install-pending-temp", "after-install-pending", "before-policy-write", "after-policy-write", "after-install-postimage", "after-install-final-temp", "after-install-final", ) for point in install_points: with self.subTest(operation="install", point=point): self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() self._kill_paused(args, "-Install", point) if point == "after-install-final": result = self._run(self.installer, *args, "-HealthCheck") self.assertEqual(result.returncode, 0, (point, result.stdout, result.stderr)) self.assertTrue(Path(fixture["paths"]["install_final"]).exists()) self.assertNotEqual(before, fixture["registry"].read_bytes()) self.assertEqual(self._failure_records(fixture), []) elif point in ("after-install-pending-temp", "after-install-final-temp"): snapshot = fixture["registry"].read_bytes() result = self._run(self.installer, *args) self.assertEqual(result.returncode, 3, (point, result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_RECEIPT") self.assertEqual(snapshot, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) else: result = self._run(self.installer, *args) self.assertEqual(result.returncode, 4, (point, result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_PROCESS_INTERRUPTED") self.assertEqual(before, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["install_recovery"]).exists()) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) rollback_points = ( "after-rollback-pending-temp", "after-rollback-pending", "before-policy-delete", "after-policy-delete", "after-rollback-postimage", "after-rollback-final-temp", "after-rollback-final", ) for point in rollback_points: with self.subTest(operation="rollback", point=point): self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() installed = self._run(self.installer, *args, "-Install") self.assertEqual(installed.returncode, 0, (installed.stdout, installed.stderr)) active = fixture["registry"].read_bytes() self._kill_paused(args, "-Rollback", point) if point == "after-rollback-final": result = self._run(self.installer, *args, "-Rollback") self.assertEqual(result.returncode, 0, (point, result.stdout, result.stderr)) self.assertTrue(Path(fixture["paths"]["rollback_final"]).exists()) self.assertEqual(before, fixture["registry"].read_bytes()) elif point in ("after-rollback-pending-temp", "after-rollback-final-temp"): snapshot = fixture["registry"].read_bytes() result = self._run(self.installer, *args, "-Rollback") self.assertEqual(result.returncode, 3, (point, result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_RECEIPT") self.assertEqual(snapshot, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["rollback_final"]).exists()) else: result = self._run(self.installer, *args, "-Rollback") self.assertEqual(result.returncode, 4, (point, result.stdout, result.stderr)) self.assertEqual(json.loads(result.stdout)["error_code"], "E_PROCESS_INTERRUPTED") self.assertEqual(active, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["rollback_recovery"]).exists()) self.assertFalse(Path(fixture["paths"]["rollback_final"]).exists()) @unittest.skipUnless(os.name == "nt" and hasattr(signal, "CTRL_BREAK_EVENT"), "Windows CTRL_BREAK_EVENT required") def test_17_actual_ctrl_break_has_no_success_and_reopen_recovers(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() stdout, _ = self._kill_paused(args, "-Install", "after-policy-write", ctrl_break=True) self.assertNotIn("INSTALL_COMPLETE", stdout) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) reopened = self._run(self.installer, *args) self.assertEqual(reopened.returncode, 4, (reopened.stdout, reopened.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertTrue(Path(fixture["paths"]["install_recovery"]).exists()) def test_18_recovery_receipt_crash_points_are_durable_and_never_retry_install(self) -> None: cases = { "before-recovery-receipt": 4, "after-recovery-receipt-temp": 3, "after-recovery-receipt": 4, } for point, expected_exit in cases.items(): with self.subTest(point=point): self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before = fixture["registry"].read_bytes() marker = self.root / (point + ".marker") process = subprocess.Popen( [POWERSHELL, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(self.installer), *args, "-Install", "-AdapterInjectFailure", "after-policy-write", "-AdapterPauseAt", point, "-AdapterPauseMarker", str(marker)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=PROJECT, ) self._wait_for_marker(marker) process.kill() process.communicate(timeout=10) reopened = self._run(self.installer, *args) self.assertEqual(reopened.returncode, expected_exit, (point, reopened.stdout, reopened.stderr)) self.assertEqual(before, fixture["registry"].read_bytes()) self.assertFalse(Path(fixture["paths"]["install_final"]).exists()) if point == "after-recovery-receipt-temp": self.assertEqual(json.loads(reopened.stdout)["error_code"], "E_RECEIPT") else: self.assertEqual(json.loads(reopened.stdout)["error_code"], "E_PROCESS_INTERRUPTED") self.assertTrue(Path(fixture["paths"]["install_recovery"]).exists()) def test_19_schema2_immutable_audit_section_and_approval_path_are_shared(self) -> None: fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) registry_before = fixture["registry"].read_bytes() approval_before = self.source_approval.read_bytes() # A later append-only audit section cannot invalidate the approved immutable section. self.audit.write_text( self.audit.read_text(encoding="utf-8") + "\n## DEV-AUDIT-UNRELATED-LATER-APPEND-20260810-001\n\n- verdict=PASS/0\n", encoding="utf-8", newline="\n", ) builder_ok = self._run(self.builder, *self._source_args()) installer_ok = self._run(self.installer, *args) self.assertEqual(builder_ok.returncode, 0, (builder_ok.stdout, builder_ok.stderr)) self.assertEqual(installer_ok.returncode, 0, (installer_ok.stdout, installer_ok.stderr)) self.assertEqual(registry_before, fixture["registry"].read_bytes()) # Schema 1 and the immutable historical path are never accepted as fallback. value = json.loads(approval_before.decode("utf-8")) value["schema"] = 1 dump_json(self.source_approval, value) for script, script_args in ((self.builder, self._source_args()), (self.installer, self._install_args(fixture))): failed = self._run(script, *script_args) self.assertEqual(failed.returncode, 3, (failed.stdout, failed.stderr)) self.source_approval.write_bytes(approval_before) self.historical_source_approval.parent.mkdir(parents=True, exist_ok=True) self.historical_source_approval.write_bytes(approval_before) historical_args = [ "-ApprovedSourceReceipt", str(self.historical_source_approval), "-ApprovedSourceReceiptBytes", str(self.historical_source_approval.stat().st_size), "-ApprovedSourceReceiptSha256", sha_file(self.historical_source_approval), ] failed_builder = self._run(self.builder, *historical_args, "-AdapterAllowTestScope") failed_installer = self._run( self.installer, *historical_args, *self._install_args(fixture)[6:], ) self.assertEqual(failed_builder.returncode, 3, (failed_builder.stdout, failed_builder.stderr)) self.assertEqual(failed_installer.returncode, 3, (failed_installer.stdout, failed_installer.stderr)) self.assertEqual(registry_before, fixture["registry"].read_bytes()) # Mutation, omission, duplicate ID and a freshly re-signed cross-source section all fail # before candidate output or policy state can change in both consumers. for mode in ("mutation", "omission", "duplicate", "cross-source"): with self.subTest(mode=mode): self._reset() fixture = self._make_install_fixture(key_exists=False) args = self._install_args(fixture) before_registry = fixture["registry"].read_bytes() approval = json.loads(self.source_approval.read_text(encoding="utf-8")) audit_id = approval["implementation_review"]["audit_id"] audit_text = self.audit.read_text(encoding="utf-8") if mode == "mutation": self.audit.write_text(audit_text.replace("- verdict=PASS/0", "- verdict=HOLD/1", 1), encoding="utf-8", newline="\n") elif mode == "omission": self.audit.write_text("## DEV-AUDIT-DIFFERENT-001\n\n- verdict=PASS/0\n", encoding="utf-8", newline="\n") elif mode == "duplicate": self.audit.write_text(audit_text + "\n" + audit_text, encoding="utf-8", newline="\n") else: self.audit.write_text(audit_text.replace(approval["managed_load_source_tree"]["tree_sha256"], "0" * 64, 1), encoding="utf-8", newline="\n") section = canonical_audit_section(self.audit, audit_id, self.contract) approval["implementation_review"]["audit_section_bytes"] = len(section) approval["implementation_review"]["audit_section_sha256"] = sha_bytes(section) dump_json(self.source_approval, approval) args = self._install_args(fixture) output = self.root / f"{mode}-candidate" failed_builder = self._run(self.builder, *self._source_args(), "-Build", "-OutputRoot", str(output)) failed_installer = self._run(self.installer, *args) self.assertEqual(failed_builder.returncode, 3, (mode, failed_builder.stdout, failed_builder.stderr)) self.assertEqual(failed_installer.returncode, 3, (mode, failed_installer.stdout, failed_installer.stderr)) self.assertFalse(output.exists()) self.assertEqual(before_registry, fixture["registry"].read_bytes()) if __name__ == "__main__": unittest.main()