from __future__ import annotations import hashlib import json import os import pathlib import tempfile import unittest from unittest import mock PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[4] PROJECT_DEV = PROJECT_ROOT / "dev" / "project-dev" if str(PROJECT_DEV) not in os.sys.path: os.sys.path.insert(0, str(PROJECT_DEV)) from bili_authenticated_extension.constants import ( # noqa: E402 EXTENSION_BUILD, HOST_BUILD, MANIFEST_PUBLIC_KEY, RELOAD_GENERATION, stable_job_id, ) from bili_authenticated_extension.protocol import ProtocolError # noqa: E402 from bili_authenticated_extension import queue_producer as producer # noqa: E402 from bili_authenticated_extension.queue_producer import ( # noqa: E402 _load_release_approval, _projection_tree, _successor_scope_sha256, ) from bili_authenticated_extension.queue_state import QueueStore # noqa: E402 CREATOR = "246813579" BVID = "BV1AbCd2EfGh" AUTH_MESSAGE = "msg_20260817123456789_deadbeef" AUTH_HANDOFF = "HANDOFF-INFOADMIN-INFODEV2-SYNTHETIC-SUCCESSOR-20260817-001" SOURCE_FILES = ( "__init__.py", "background.js", "build_host.ps1", "config.example.json", "constants.py", "dependencies/dependency-artifact-manifest.json", "dependencies/yt_dlp-2026.7.4-py3-none-any.whl", "formal_legacy_identity_manifest.py", "install_native_host.ps1", "job.py", "manifest.json", "native-host-manifest.template.json", "native_host.py", "protocol.py", "queue-producer.example.json", "queue_producer.py", "queue_state.py", "sidepanel.css", "sidepanel.html", "sidepanel.js", "worker.py", ) def encode(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" def write_json(path: pathlib.Path, value: object) -> bytes: path.parent.mkdir(parents=True, exist_ok=True) payload = encode(value) path.write_bytes(payload) return payload def file_spec(root: pathlib.Path, path: pathlib.Path) -> dict[str, object]: payload = path.read_bytes() return { "relative_path": path.relative_to(root).as_posix(), "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper(), } def build_environment( root: pathlib.Path, *, fast_path: bool = False, fast_path_mode: str = "PRODUCER_ONLY_DELTA", ) -> tuple[dict[str, object], pathlib.Path, pathlib.Path, pathlib.Path]: source_root = root / "dev/project-dev/bili_authenticated_extension" projection_root = root / "dev/project-dev/bili_authenticated_extension_unpacked" runtime = root / "runtime" runtime.mkdir(parents=True) queue = runtime / "queue.jsonl" state = runtime / "state.jsonl" lock = runtime / "queue.lock" reload = runtime / "reload.jsonl" lock.write_bytes(b"\0") manifest = { "manifest_version": 3, "name": "synthetic generic", "version": "1.2.25", "version_name": "1.2.25+20260829.generic.v027", "key": MANIFEST_PUBLIC_KEY, } payloads: dict[str, bytes] = { "background.js": f'const EXTENSION_BUILD = "{EXTENSION_BUILD}";\n'.encode("ascii"), "manifest.json": encode(manifest), "sidepanel.css": b"body{}\n", "sidepanel.html": b"synthetic\n", "sidepanel.js": b"export {};\n", } for name in SOURCE_FILES: payloads.setdefault(name, f"synthetic:{name}\n".encode("ascii")) dependency_payload = b'{"schema":1,"synthetic":true}\n' payloads["dependencies/dependency-artifact-manifest.json"] = dependency_payload entries = [] for name in ("background.js", "manifest.json", "sidepanel.css", "sidepanel.html", "sidepanel.js"): payload = payloads[name] for parent in (source_root, projection_root): (parent / name).parent.mkdir(parents=True, exist_ok=True) (parent / name).write_bytes(payload) entries.append({ "path": name, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper(), }) source_entries = [] for name in SOURCE_FILES: payload = payloads[name] candidate = source_root / name candidate.parent.mkdir(parents=True, exist_ok=True) candidate.write_bytes(payload) source_entries.append({ "path": name, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper(), }) tree_hash = _projection_tree(entries) contract_payload = write_json(root / "dev/project-dev/bili_authenticated_extension_unpacked_contract.json", { "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "project_id": "project-info", "source_root": "dev/project-dev/bili_authenticated_extension", "projection_root": "dev/project-dev/bili_authenticated_extension_unpacked", "expected_extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "public_key_der_sha256": "E83C2B2AF3CF011543FBA13FBC524D1122EEA68548F9F27B9F7A82B5D594666C", "tree_hash_algorithm": "path-nul-bytes-nul-sha256-upper-lf-v1", "tree_sha256": tree_hash, "files": sorted(entries, key=lambda item: item["path"]), }) source_manifest_value = { "schema": 1, "scope": "generic-bilibili-queue", "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "extension_build": EXTENSION_BUILD, "host_build": HOST_BUILD, "archive_metadata_contract": { "schema": 1, "root": "yt_dlp-2026.7.4.dist-info", "relative_files": ["INSTALLER", "METADATA", "RECORD", "REQUESTED", "WHEEL", "entry_points.txt", "licenses/LICENSE"], "distribution_name": "yt-dlp", "distribution_version": "2026.7.4", "allowed_type_codes": ["b", "x"], "source_date_epoch": 1786207924, "tree_hash_algorithm": "sha256(path-utf8,nul,decimal-bytes-ascii,nul,payload-sha256-lower-hex-ascii,lf)-upper-hex-v1", "canonical_tree_sha256": "32F1DC23F6704966AE4511CB173318CD11FCBA031323028D268D9F2488589E70", }, "dependency_artifact_manifest_bytes": len(dependency_payload), "dependency_artifact_manifest_sha256": hashlib.sha256(dependency_payload).hexdigest().upper(), "files": source_entries, } source_manifest_payload = write_json( source_root / "source-artifact-manifest.json", source_manifest_value, ) source_hash = hashlib.sha256(source_manifest_payload).hexdigest().upper() host_build_source_manifest_payload = source_manifest_payload host_build_source_manifest_hash = source_hash host_build_source_manifest_path = root / "ai-infoadmin/worklog/synthetic-fast-path-host-build-source-manifest.json" if fast_path: host_source = json.loads(json.dumps(source_manifest_value)) if fast_path_mode == "PRODUCER_ONLY_DELTA": for item in host_source["files"]: if item["path"] == "queue_producer.py": previous = b"synthetic:queue_producer.py:previous\n" item["bytes"] = len(previous) item["sha256"] = hashlib.sha256(previous).hexdigest().upper() break host_build_source_manifest_payload = write_json(host_build_source_manifest_path, host_source) host_build_source_manifest_hash = hashlib.sha256(host_build_source_manifest_payload).hexdigest().upper() build_root = root / "dev/tmp/synthetic-build" exe = build_root / "project-info-bili-auth-native-host.exe" exe.parent.mkdir(parents=True) exe.write_bytes(b"MZsynthetic-v002") exe_hash = hashlib.sha256(exe.read_bytes()).hexdigest().upper() build_receipt_payload = write_json(build_root / "build-artifact-manifest.json", { "schema": 2, "scope": "generic-bilibili-queue", "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "extension_build": EXTENSION_BUILD, "host_build": HOST_BUILD, "packaging": "pyinstaller-onefile", "pyinstaller_version": "6.15.0", "yt_dlp_version": "2026.7.4", "builder_python_sha256": "A" * 64, "pyinstaller_executable_bytes": 1, "pyinstaller_executable_sha256": "B" * 64, "builder_provision_receipt_bytes": 1, "builder_provision_receipt_sha256": "C" * 64, "build_script_sha256": "D" * 64, "source_artifact_manifest_bytes": len(host_build_source_manifest_payload), "source_artifact_manifest_sha256": host_build_source_manifest_hash, "dependency_artifact_manifest_bytes": len(dependency_payload), "dependency_artifact_manifest_sha256": hashlib.sha256(dependency_payload).hexdigest().upper(), "yt_dlp_wheel_sha256": "E" * 64, "archive_verification": { "status": "PASS", "method": "synthetic static archive", "required_modules": ["bili_authenticated_extension.worker", "yt_dlp", "yt_dlp.downloader", "yt_dlp.globals", "yt_dlp.plugins", "yt_dlp.version"], "metadata_entry": "yt_dlp-2026.7.4.dist-info/METADATA", "metadata_files": 7, "metadata_type_codes": ["b"], "metadata_tree_sha256": "32F1DC23F6704966AE4511CB173318CD11FCBA031323028D268D9F2488589E70", }, "files": [{"path": exe.name, "bytes": exe.stat().st_size, "sha256": exe_hash}], }) source_receipt_path = root / ( "ai-infoadmin/worklog/synthetic-fast-path-source-receipt.json" if fast_path else "ai-inforev/worklog/synthetic-source-approval.json" ) build_approval_payload = write_json(source_receipt_path, { "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "approval_scope": "controlled-build-source-manifest", "approved_by_role": "project.admin" if fast_path else "dev.reviewer.project", "status": "APPROVED", "source_artifact_manifest_bytes": len(source_manifest_payload), "source_artifact_manifest_sha256": source_hash, }) build_validation_receipt_path = root / ( "ai-infoadmin/worklog/synthetic-fast-path-build-receipt.json" if fast_path else "ai-inforev/worklog/synthetic-install-approval.json" ) install_approval_payload = write_json(build_validation_receipt_path, { "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "approval_scope": "install-exact-build", "approved_by_role": "project.admin" if fast_path else "dev.reviewer.project", "status": "APPROVED", "source_artifact_manifest_sha256": host_build_source_manifest_hash, "build_artifact_manifest_bytes": len(build_receipt_payload), "build_artifact_manifest_sha256": hashlib.sha256(build_receipt_payload).hexdigest().upper(), "host_executable_bytes": len(exe.read_bytes()), "host_executable_sha256": exe_hash, }) installed_root = root / "installed/generic-v002" installed_root.mkdir(parents=True) installed_exe = installed_root / exe.name installed_exe.write_bytes(exe.read_bytes()) host_config = installed_root / "config.json" host_config_payload = write_json(host_config, { "schema": 2, "creator_allowlist": [CREATOR], "queue_path": str(queue), "queue_state_path": str(state), "queue_lock_path": str(lock), "reload_state_path": str(reload), "reload_generation": RELOAD_GENERATION, "required_extension_build": EXTENSION_BUILD, "ffmpeg": str(root / "tools/ffmpeg.exe"), "ffmpeg_sha256": "1" * 64, "ffprobe": str(root / "tools/ffprobe.exe"), "ffprobe_sha256": "2" * 64, "bridge_python": str(root / "tools/python.exe"), "bridge_python_sha256": "3" * 64, "bridge_script": str(root / "tools/bridge.py"), "bridge_script_sha256": "4" * 64, "yt_dlp_executable": str(root / "tools/yt-dlp.exe"), "yt_dlp_executable_sha256": "5" * 64, "destination": str(root / "destination"), "creator_name": "Synthetic Creator", "formal_manifest_path": str(root / "formal/manifest.jsonl"), "processing_handoff_path": str(root / "formal/video-processing-handoffs.jsonl"), }) native_manifest = installed_root / "native-host-manifest.json" native_manifest_payload = write_json(native_manifest, { "name": "com.project_info.bili_auth_ingress", "description": "synthetic", "path": str(installed_exe), "type": "stdio", "allowed_origins": ["chrome-extension://oidmclckpdmpabbfedplkbdplmfcenbb/"], }) installed_files = [] for candidate in (installed_exe, host_config, native_manifest): payload = candidate.read_bytes() installed_files.append({ "path": candidate.name, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper(), }) install_receipt_path = root / ( "ai-infoadmin/worklog/synthetic-fast-path-install-readiness-receipt.json" if fast_path else "ai-infoadmin/worklog/synthetic-install-receipt.json" ) if not fast_path: install_receipt_payload = write_json(install_receipt_path, { "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "host_build": HOST_BUILD, "required_extension_build": EXTENSION_BUILD, "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "host_name": "com.project_info.bili_auth_ingress", "installed_root": str(installed_root), "installed_files": installed_files, }) write_json(reload, { "schema": 1, "generation": RELOAD_GENERATION, "event": "APPLIED", "token": "a" * 32, "from_build": EXTENSION_BUILD, "to_build": EXTENSION_BUILD, "at_unix_ms": 2_000_000_000_000, }) ingress = { "schema": 1, "bvid": BVID, "creator_uid": CREATOR, "expected_duration_ms": 180_000, "discovered_at_unix_ms": 1_900_000_000_000, "published_at": "2026-08-17T12:00:00+08:00", "title": "synthetic title", } queue.write_bytes(encode(ingress)) store = QueueStore(queue, state, lock, frozenset({CREATOR})) claimed = store.claim_next(2_000_000_000_100) assert claimed is not None store.reject_claim(claimed[0], claimed[1], 2_000_000_000_101, "E_PAGE_PROOF", { "attempts": 1, "elapsed_ms": 1, "state": "PAGE_REJECTED", "reason": "PAGE_IDENTITY_REJECTED", }) repair_audit_id = "DEV-AUDIT-PROJECT-INFO-SYNTHETIC-PAGE-PROOF-20260817-001" implementation_audit_id = "DEV-AUDIT-PROJECT-INFO-SYNTHETIC-LINEAGE-IMPLEMENTATION-20260817-001" repair_prefix = f"# audit\n\n## {repair_audit_id}\nPASS/0\n".encode("utf-8") implementation_prefix = repair_prefix + f"\n## {implementation_audit_id}\nPASS/0\n".encode("utf-8") audit_path = root / "dev-doc/开发审计报告.md" audit_path.parent.mkdir(parents=True, exist_ok=True) audit_path.write_bytes(implementation_prefix) repair = { "review_result_message_id": "msg_20260817111111111_a1b2c3d4", "audit_id": repair_audit_id, "audit_bytes": len(repair_prefix), "audit_sha256": hashlib.sha256(repair_prefix).hexdigest().upper(), "verdict": "PASS/0", "blocking_findings": 0, } authorization_path = root / "ai-infoadmin/worklog/synthetic-authorization.json" authorization_value = { "schema": 1, "scope": "bili-auth-successor-lineage-v1", "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "authorized_by_role": "project.admin", "authorization_message_id": AUTH_MESSAGE, "authorization_handoff_id": AUTH_HANDOFF, "repair": repair, "successors": [{ "creator_uid": CREATOR, "bvid": BVID, "predecessor_job_id": stable_job_id(CREATOR, BVID), "retry_generation": 1, "terminal_error_code": "E_PAGE_PROOF", }], } authorization_payload = write_json(authorization_path, authorization_value) review = { "result_message_id": "msg_20260817122222222_11223344", "audit_id": implementation_audit_id, "audit_bytes": len(implementation_prefix), "audit_sha256": hashlib.sha256(implementation_prefix).hexdigest().upper(), "verdict": "PASS/0", "blocking_findings": 0, } formal = root / "formal/manifest.jsonl" formal.parent.mkdir(parents=True, exist_ok=True) formal.write_bytes(b'{"schema":1,"synthetic":true}\n') if fast_path: def frozen_fields(path: pathlib.Path, prefix: str, *, exact: bool) -> dict[str, object]: payload = path.read_bytes() size_name = f"{prefix}_bytes" if exact else f"{prefix}_prefix_bytes" lines_name = f"{prefix}_lines" if exact else f"{prefix}_prefix_lines" sha_name = f"{prefix}_sha256" if exact else f"{prefix}_prefix_sha256" return { f"{prefix}_path": str(path), size_name: len(payload), lines_name: payload.count(b"\n"), sha_name: hashlib.sha256(payload).hexdigest().upper(), } install_receipt = { "schema": 2, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "validation_scope": "continuous-fast-path-installed-readiness-v1", "validated_by_role": "project.admin", "status": "VALIDATED", "continuous_authorization_handoff_id": producer._CONTINUOUS_FAST_PATH_HANDOFF_ID, "owner_ai_id": "infodev-2", "owner_thread_id": "019fbcbb-bed7-7c90-83ab-f50610f80d3a", "owner_role_instance_id": "dev.developer.project.secondary", "authorization_file_relative_path": authorization_path.relative_to(root).as_posix(), "authorization_file_bytes": len(authorization_payload), "authorization_file_sha256": hashlib.sha256(authorization_payload).hexdigest().upper(), "successor_scope_sha256": _successor_scope_sha256(authorization_value["successors"]), "implementation_review_audit_id": review["audit_id"], "implementation_review_audit_bytes": review["audit_bytes"], "implementation_review_audit_sha256": review["audit_sha256"], "source_artifact_manifest_bytes": len(source_manifest_payload), "source_artifact_manifest_sha256": source_hash, "source_receipt_bytes": len(build_approval_payload), "source_receipt_sha256": hashlib.sha256(build_approval_payload).hexdigest().upper(), "build_artifact_manifest_bytes": len(build_receipt_payload), "build_artifact_manifest_sha256": hashlib.sha256(build_receipt_payload).hexdigest().upper(), "build_receipt_bytes": len(install_approval_payload), "build_receipt_sha256": hashlib.sha256(install_approval_payload).hexdigest().upper(), "host_executable_bytes": len(exe.read_bytes()), "host_executable_sha256": exe_hash, "host_build": HOST_BUILD, "required_extension_build": EXTENSION_BUILD, "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "host_name": "com.project_info.bili_auth_ingress", "installed_root": str(installed_root), "installed_files": installed_files, "native_messaging_host_manifest": str(native_manifest), "host_build_source_manifest": file_spec(root, host_build_source_manifest_path), "host_source_binding_mode": fast_path_mode, "producer_only_changed_files": ["queue_producer.py"] if fast_path_mode == "PRODUCER_ONLY_DELTA" else [], "host_archive_excluded_modules": ["bili_authenticated_extension.queue_producer"], "projection_contract_sha256": hashlib.sha256(contract_payload).hexdigest().upper(), "projection_tree_sha256": tree_hash, "secret_field_count": 0, } install_receipt.update(frozen_fields(reload, "reload_state", exact=True)) install_receipt.update(frozen_fields(queue, "queue", exact=False)) install_receipt.update(frozen_fields(state, "queue_state", exact=False)) install_receipt.update(frozen_fields(formal, "formal_manifest", exact=False)) install_receipt_payload = write_json(install_receipt_path, install_receipt) deployment = { "extension_build": EXTENSION_BUILD, "host_build": HOST_BUILD, "reload_generation": RELOAD_GENERATION, "source_manifest_sha256": source_hash, "build_approval": file_spec(root, source_receipt_path), "build_receipt": file_spec(root, build_root / "build-artifact-manifest.json"), "exe": file_spec(root, exe), "install_approval": file_spec(root, build_validation_receipt_path), "install_receipt": file_spec(root, install_receipt_path), "installed_config_sha256": hashlib.sha256(host_config_payload).hexdigest().upper(), "installed_manifest_sha256": hashlib.sha256(native_manifest_payload).hexdigest().upper(), "installed_exe_sha256": exe_hash, "projection_contract_sha256": hashlib.sha256(contract_payload).hexdigest().upper(), "projection_tree_sha256": tree_hash, "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", } write_json(root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json", { "schema": 1, "scope": "bili-auth-successor-exact-release-v1", "project_id": "project-info", "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "approved_by_role": "project.admin", "authorization_message_id": AUTH_MESSAGE, "authorization_handoff_id": AUTH_HANDOFF, "authorization_file": file_spec(root, authorization_path), "implementation_review": review, "deployment": deployment, }) config = { "schema": 2, "project_root": root, "host_config_path": host_config, "creator_uid": CREATOR, "successor_authorization_message_id": AUTH_MESSAGE, "queue_paths": { "queue_path": queue, "queue_state_path": state, "queue_lock_path": lock, "reload_state_path": reload, }, "_host_config_sha256": hashlib.sha256(host_config_payload).hexdigest().upper(), } return config, queue, state, native_manifest def refresh_installed_identity(root: pathlib.Path, name: str) -> None: installed_root = root / "installed/generic-v002" receipt_path = root / "ai-infoadmin/worklog/synthetic-install-receipt.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) candidate = installed_root / name payload = candidate.read_bytes() for item in receipt["installed_files"]: if item["path"] == name: item["bytes"] = len(payload) item["sha256"] = hashlib.sha256(payload).hexdigest().upper() break write_json(receipt_path, receipt) approval_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" approval = json.loads(approval_path.read_text(encoding="utf-8")) approval["deployment"]["install_receipt"] = file_spec(root, receipt_path) field = { "config.json": "installed_config_sha256", "project-info-bili-auth-native-host.exe": "installed_exe_sha256", "native-host-manifest.json": "installed_manifest_sha256", }[name] approval["deployment"][field] = hashlib.sha256(payload).hexdigest().upper() write_json(approval_path, approval) def rebind_deployment_file(root: pathlib.Path, field: str, path: pathlib.Path) -> None: approval_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" approval = json.loads(approval_path.read_text(encoding="utf-8")) approval["deployment"][field] = file_spec(root, path) write_json(approval_path, approval) def rewrite_fast_receipt(root: pathlib.Path, value: dict[str, object]) -> pathlib.Path: path = root / "ai-infoadmin/worklog/synthetic-fast-path-install-readiness-receipt.json" write_json(path, value) rebind_deployment_file(root, "install_receipt", path) return path def assert_deployment_rejected( testcase: unittest.TestCase, root: pathlib.Path, config: dict[str, object], queue: pathlib.Path, state: pathlib.Path, native_manifest: pathlib.Path, ) -> None: frozen = (queue.read_bytes(), state.read_bytes()) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_exact", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._fast_path_canonical_paths", return_value={ "queue": queue, "queue_state": state, "reload_state": config["queue_paths"]["reload_state_path"], "formal_manifest": root / "formal/manifest.jsonl", }), testcase.assertRaisesRegex(ProtocolError, "E_DEPLOYMENT_NOT_READY"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) testcase.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) class SuccessorTrustGateTests(unittest.TestCase): def test_exact_release_and_v002_deployment_pass_before_append(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) before_queue, before_state = queue.read_bytes(), state.read_bytes() ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), ): result = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) replay = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual({"appended": 1, "unchanged": 0, "recovered": 0}, result) self.assertEqual({"appended": 0, "unchanged": 1, "recovered": 0}, replay) self.assertTrue(queue.read_bytes().startswith(before_queue)) self.assertEqual(before_state, state.read_bytes()) def test_exact_project_admin_continuous_fast_path_passes_and_replays(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root, fast_path=True) before_queue, before_state = queue.read_bytes(), state.read_bytes() ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_exact", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._fast_path_canonical_paths", return_value={ "queue": queue, "queue_state": state, "reload_state": config["queue_paths"]["reload_state_path"], "formal_manifest": root / "formal/manifest.jsonl", }), ): result = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) replay = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual({"appended": 1, "unchanged": 0, "recovered": 0}, result) self.assertEqual({"appended": 0, "unchanged": 1, "recovered": 0}, replay) self.assertTrue(queue.read_bytes().startswith(before_queue)) self.assertEqual(before_state, state.read_bytes()) def test_project_admin_same_source_host_build_passes_and_replays(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment( root, fast_path=True, fast_path_mode="SAME_SOURCE_HOST_BUILD", ) before_queue, before_state = queue.read_bytes(), state.read_bytes() ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_exact", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._fast_path_canonical_paths", return_value={ "queue": queue, "queue_state": state, "reload_state": config["queue_paths"]["reload_state_path"], "formal_manifest": root / "formal/manifest.jsonl", }), ): result = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) replay = store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual({"appended": 1, "unchanged": 0, "recovered": 0}, result) self.assertEqual({"appended": 0, "unchanged": 1, "recovered": 0}, replay) self.assertTrue(queue.read_bytes().startswith(before_queue)) self.assertEqual(before_state, state.read_bytes()) def test_project_admin_fast_path_spoof_drift_and_cross_branch_matrix_is_mutation_zero(self) -> None: mutations = ( "source-path-shadow", "cross-branch-mix", "wrong-source-role", "wrong-source-task", "wrong-continuous-authorization", "wrong-scope", "wrong-status", "wrong-schema", "wrong-owner", "wrong-successor-scope", "changed-authorization-generation", "extra-receipt-field", "drift-source-binding", "drift-build-binding", "drift-installed-binding", "drift-readiness", "drift-queue-prefix", "drift-state-prefix", "drift-formal-prefix", "shadow-queue-path", "nonzero-secret-field-count", "host-source-nonproducer-delta", "wrong-host-source-mode", "same-source-declared-for-delta", ) for mutation in mutations: with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root, fast_path=True) receipt_path = root / "ai-infoadmin/worklog/synthetic-fast-path-install-readiness-receipt.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) if mutation == "source-path-shadow": source = root / "ai-infoadmin/worklog/synthetic-fast-path-source-receipt.json" shadow = root / "ai-infoadmin/worklog/shadow/equal-fast-path-source-receipt.json" shadow.parent.mkdir(parents=True) shadow.write_bytes(source.read_bytes()) rebind_deployment_file(root, "build_approval", shadow) elif mutation == "cross-branch-mix": source = root / "ai-infoadmin/worklog/synthetic-fast-path-build-receipt.json" mixed = root / "ai-inforev/worklog/synthetic-install-approval.json" mixed.parent.mkdir(parents=True, exist_ok=True) mixed.write_bytes(source.read_bytes()) rebind_deployment_file(root, "install_approval", mixed) elif mutation in {"wrong-source-role", "wrong-source-task"}: source = root / "ai-infoadmin/worklog/synthetic-fast-path-source-receipt.json" source_value = json.loads(source.read_text(encoding="utf-8")) if mutation == "wrong-source-role": source_value["approved_by_role"] = "dev.reviewer.project" else: source_value["task_id"] = "DEV-ANOTHER-TASK" source_payload = write_json(source, source_value) receipt["source_receipt_bytes"] = len(source_payload) receipt["source_receipt_sha256"] = hashlib.sha256(source_payload).hexdigest().upper() rewrite_fast_receipt(root, receipt) rebind_deployment_file(root, "build_approval", source) elif mutation == "changed-authorization-generation": authorization_path = root / "ai-infoadmin/worklog/synthetic-authorization.json" authorization = json.loads(authorization_path.read_text(encoding="utf-8")) authorization["successors"][0]["retry_generation"] = 2 authorization_payload = write_json(authorization_path, authorization) release_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" release = json.loads(release_path.read_text(encoding="utf-8")) release["authorization_file"] = file_spec(root, authorization_path) receipt["authorization_file_bytes"] = len(authorization_payload) receipt["authorization_file_sha256"] = hashlib.sha256(authorization_payload).hexdigest().upper() write_json(release_path, release) rewrite_fast_receipt(root, receipt) elif mutation == "host-source-nonproducer-delta": host_source_path = root / "ai-infoadmin/worklog/synthetic-fast-path-host-build-source-manifest.json" host_source = json.loads(host_source_path.read_text(encoding="utf-8")) for item in host_source["files"]: if item["path"] == "worker.py": item["sha256"] = "2" * 64 break host_source_payload = write_json(host_source_path, host_source) host_source_hash = hashlib.sha256(host_source_payload).hexdigest().upper() artifact_path = root / "dev/tmp/synthetic-build/build-artifact-manifest.json" artifact = json.loads(artifact_path.read_text(encoding="utf-8")) artifact["source_artifact_manifest_bytes"] = len(host_source_payload) artifact["source_artifact_manifest_sha256"] = host_source_hash artifact_payload = write_json(artifact_path, artifact) build_path = root / "ai-infoadmin/worklog/synthetic-fast-path-build-receipt.json" build_value = json.loads(build_path.read_text(encoding="utf-8")) build_value["source_artifact_manifest_sha256"] = host_source_hash build_value["build_artifact_manifest_bytes"] = len(artifact_payload) build_value["build_artifact_manifest_sha256"] = hashlib.sha256(artifact_payload).hexdigest().upper() build_payload = write_json(build_path, build_value) receipt["host_build_source_manifest"] = file_spec(root, host_source_path) receipt["build_artifact_manifest_bytes"] = len(artifact_payload) receipt["build_artifact_manifest_sha256"] = hashlib.sha256(artifact_payload).hexdigest().upper() receipt["build_receipt_bytes"] = len(build_payload) receipt["build_receipt_sha256"] = hashlib.sha256(build_payload).hexdigest().upper() rewrite_fast_receipt(root, receipt) rebind_deployment_file(root, "build_receipt", artifact_path) rebind_deployment_file(root, "install_approval", build_path) else: if mutation == "wrong-continuous-authorization": receipt["continuous_authorization_handoff_id"] = "HANDOFF-OTHER-CONTINUOUS-AUTH" elif mutation == "wrong-scope": receipt["validation_scope"] = "other-scope" elif mutation == "wrong-status": receipt["status"] = "PENDING" elif mutation == "wrong-schema": receipt["schema"] = 1 elif mutation == "wrong-owner": receipt["owner_thread_id"] = "01900000-0000-0000-0000-000000000000" elif mutation == "wrong-successor-scope": receipt["successor_scope_sha256"] = "A" * 64 elif mutation == "extra-receipt-field": receipt["unexpected"] = "forbidden" elif mutation == "drift-source-binding": receipt["source_artifact_manifest_sha256"] = "B" * 64 elif mutation == "drift-build-binding": receipt["build_artifact_manifest_sha256"] = "C" * 64 elif mutation == "drift-installed-binding": receipt["installed_files"][0]["sha256"] = "D" * 64 elif mutation == "drift-readiness": receipt["reload_state_sha256"] = "E" * 64 elif mutation == "drift-queue-prefix": receipt["queue_prefix_sha256"] = "F" * 64 elif mutation == "drift-state-prefix": receipt["queue_state_prefix_sha256"] = "0" * 64 elif mutation == "drift-formal-prefix": receipt["formal_manifest_prefix_sha256"] = "1" * 64 elif mutation == "shadow-queue-path": shadow = root / "shadow/queue.jsonl" shadow.parent.mkdir(parents=True) shadow.write_bytes(queue.read_bytes()) receipt["queue_path"] = str(shadow) elif mutation == "nonzero-secret-field-count": receipt["secret_field_count"] = 1 elif mutation == "wrong-host-source-mode": receipt["host_source_binding_mode"] = "UNRESTRICTED" elif mutation == "same-source-declared-for-delta": receipt["host_source_binding_mode"] = "SAME_SOURCE_HOST_BUILD" receipt["producer_only_changed_files"] = [] rewrite_fast_receipt(root, receipt) assert_deployment_rejected(self, root, config, queue, state, native_manifest) def test_producer_only_delta_declared_for_same_source_is_mutation_zero(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment( root, fast_path=True, fast_path_mode="SAME_SOURCE_HOST_BUILD", ) receipt_path = root / "ai-infoadmin/worklog/synthetic-fast-path-install-readiness-receipt.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) receipt["host_source_binding_mode"] = "PRODUCER_ONLY_DELTA" receipt["producer_only_changed_files"] = ["queue_producer.py"] rewrite_fast_receipt(root, receipt) assert_deployment_rejected(self, root, config, queue, state, native_manifest) def test_project_admin_fast_path_exact_registry_gate_rejects_extra_state_before_append(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root, fast_path=True) frozen = (queue.read_bytes(), state.read_bytes()) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_exact", side_effect=ProtocolError("E_DEPLOYMENT_NOT_READY")), mock.patch("bili_authenticated_extension.queue_producer._fast_path_canonical_paths", return_value={ "queue": queue, "queue_state": state, "reload_state": config["queue_paths"]["reload_state_path"], "formal_manifest": root / "formal/manifest.jsonl", }), self.assertRaisesRegex(ProtocolError, "E_DEPLOYMENT_NOT_READY"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) def test_project_admin_fast_path_reparse_receipt_is_rejected_before_append(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root, fast_path=True) source_receipt = root / "ai-infoadmin/worklog/synthetic-fast-path-source-receipt.json" real_lstat = os.lstat class ReparseStat: def __init__(self, value: os.stat_result) -> None: self._value = value self.st_file_attributes = 0x400 def __getattr__(self, name: str) -> object: return getattr(self._value, name) def marked_lstat(path: os.PathLike[str] | str) -> os.stat_result | ReparseStat: value = real_lstat(path) if pathlib.Path(path) == source_receipt: return ReparseStat(value) return value frozen = (queue.read_bytes(), state.read_bytes()) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer.os.lstat", side_effect=marked_lstat), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), self.assertRaisesRegex(ProtocolError, "E_DEPLOYMENT_NOT_READY"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) def test_missing_fixed_approval_and_caller_override_are_not_trust_seams(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) approval_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" approval_path.rename(approval_path.with_suffix(".absent")) frozen = (queue.read_bytes(), state.read_bytes()) config["approval_path"] = str(approval_path.with_suffix(".absent")) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), self.assertRaisesRegex(ProtocolError, "E_AUTH_TRUST"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) source = (PROJECT_DEV / "bili_authenticated_extension" / "queue_producer.py").read_text(encoding="utf-8") self.assertNotIn("--approval-path", source) self.assertNotIn("--approval-hash", source) self.assertNotIn("--trust-root", source) def test_mixed_versions_reload_pending_registry_and_projection_drift_fail_before_begin(self) -> None: mutations = ("host-build", "reload-begin", "registry", "projection") for mutation in mutations: with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) if mutation == "host-build": receipt = root / "ai-infoadmin/worklog/synthetic-install-receipt.json" raw = json.loads(receipt.read_text(encoding="utf-8")) raw["host_build"] = "project-info-bili-auth-native-host/1.1.0+20260816.generic.v001" write_json(receipt, raw) elif mutation == "reload-begin": reload = config["queue_paths"]["reload_state_path"] write_json(reload, { "schema": 1, "generation": RELOAD_GENERATION, "event": "BEGIN", "token": "a" * 32, "from_build": EXTENSION_BUILD, "to_build": EXTENSION_BUILD, "at_unix_ms": 2_000_000_000_000, }) elif mutation == "projection": (root / "dev/project-dev/bili_authenticated_extension_unpacked/background.js").write_text("drift\n", encoding="utf-8") registry = str(native_manifest) if mutation != "registry" else str(root / "wrong.json") frozen = (queue.read_bytes(), state.read_bytes()) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=registry), self.assertRaisesRegex(ProtocolError, "E_DEPLOYMENT_NOT_READY"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) def test_projection_manifest_semantic_version_must_match_v024_before_begin(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, _native_manifest = build_environment(root) source_manifest_path = root / "dev/project-dev/bili_authenticated_extension/manifest.json" projection_manifest_path = root / "dev/project-dev/bili_authenticated_extension_unpacked/manifest.json" manifest = json.loads(source_manifest_path.read_text(encoding="utf-8")) manifest["version"] = "1.2.15" manifest_payload = encode(manifest) source_manifest_path.write_bytes(manifest_payload) projection_manifest_path.write_bytes(manifest_payload) contract_path = root / "dev/project-dev/bili_authenticated_extension_unpacked_contract.json" contract = json.loads(contract_path.read_text(encoding="utf-8")) for item in contract["files"]: if item["path"] == "manifest.json": item["bytes"] = len(manifest_payload) item["sha256"] = hashlib.sha256(manifest_payload).hexdigest().upper() contract["tree_sha256"] = _projection_tree(contract["files"]) contract_payload = write_json(contract_path, contract) deployment = { "projection_contract_sha256": hashlib.sha256(contract_payload).hexdigest().upper(), "projection_tree_sha256": contract["tree_sha256"], } frozen = (queue.read_bytes(), state.read_bytes()) with self.assertRaisesRegex(ProtocolError, "E_DEPLOYMENT_NOT_READY"): producer._validate_projection(root, deployment) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) def test_admin_role_authorization_hash_and_audit_prefix_tamper_fail_trust(self) -> None: for mutation in ("role", "authorization", "audit"): with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) approval_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" if mutation == "role": value = json.loads(approval_path.read_text(encoding="utf-8")) value["approved_by_role"] = "dev.developer.project.secondary" write_json(approval_path, value) elif mutation == "authorization": authorization_path = root / "ai-infoadmin/worklog/synthetic-authorization.json" authorization_path.write_bytes(authorization_path.read_bytes() + b" ") else: audit_path = root / "dev-doc/开发审计报告.md" payload = bytearray(audit_path.read_bytes()) payload[0] ^= 1 audit_path.write_bytes(payload) ingress = json.loads(queue.read_bytes().splitlines()[0]) store = QueueStore(queue, state, config["queue_paths"]["queue_lock_path"], frozenset({CREATOR})) frozen = (queue.read_bytes(), state.read_bytes()) with ( mock.patch("bili_authenticated_extension.queue_producer._project_root_from_source", return_value=root), mock.patch("bili_authenticated_extension.queue_producer._read_native_host_registry_default", return_value=str(native_manifest)), self.assertRaisesRegex(ProtocolError, "E_AUTH_TRUST"), ): store.append_authorized_successors(lambda: _load_release_approval(config), [ingress]) self.assertEqual(frozen, (queue.read_bytes(), state.read_bytes())) def test_actual_source_and_projection_exact_sets_reject_unmanifested_files_before_begin(self) -> None: mutations = ( "dev/project-dev/bili_authenticated_extension/unreviewed.py", "dev/project-dev/bili_authenticated_extension_unpacked/unreviewed.js", ) for relative in mutations: with self.subTest(relative=relative), tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) candidate = root / relative candidate.parent.mkdir(parents=True, exist_ok=True) candidate.write_bytes(b"unreviewed\n") assert_deployment_rejected(self, root, config, queue, state, native_manifest) def test_different_installed_exe_is_rejected_even_when_receipt_and_release_self_agree(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) installed_exe = root / "installed/generic-v002/project-info-bili-auth-native-host.exe" installed_exe.write_bytes(b"MZdifferent-installed-executable") refresh_installed_identity(root, installed_exe.name) assert_deployment_rejected(self, root, config, queue, state, native_manifest) def test_host_config_a_b_swap_after_initial_snapshot_is_rejected_before_begin(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) host_config = pathlib.Path(config["host_config_path"]) replacement = json.loads(host_config.read_text(encoding="utf-8")) replacement["destination"] = str(root / "other-destination") write_json(host_config, replacement) refresh_installed_identity(root, host_config.name) assert_deployment_rejected(self, root, config, queue, state, native_manifest) def test_build_receipt_schema_drift_is_rejected_even_when_outer_hashes_are_rebound(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config, queue, state, native_manifest = build_environment(root) build_receipt_path = root / "dev/tmp/synthetic-build/build-artifact-manifest.json" build_receipt = json.loads(build_receipt_path.read_text(encoding="utf-8")) build_receipt["unexpected"] = "schema-drift" build_payload = write_json(build_receipt_path, build_receipt) install_approval_path = root / "ai-inforev/worklog/synthetic-install-approval.json" install_approval = json.loads(install_approval_path.read_text(encoding="utf-8")) install_approval["build_artifact_manifest_bytes"] = len(build_payload) install_approval["build_artifact_manifest_sha256"] = hashlib.sha256(build_payload).hexdigest().upper() write_json(install_approval_path, install_approval) release_path = root / f"ai-infoadmin/worklog/bili-auth-successor-release-approval-{AUTH_MESSAGE}.json" release = json.loads(release_path.read_text(encoding="utf-8")) release["deployment"]["build_receipt"] = file_spec(root, build_receipt_path) release["deployment"]["install_approval"] = file_spec(root, install_approval_path) write_json(release_path, release) assert_deployment_rejected(self, root, config, queue, state, native_manifest) if __name__ == "__main__": unittest.main()