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