from __future__ import annotations import hashlib import json import pathlib import subprocess import sys import tempfile import time import unittest import io import struct import importlib.util import os import shutil import zipfile from unittest import mock os.environ["PYTHONDONTWRITEBYTECODE"] = "1" PROJECT_DEV = pathlib.Path(__file__).resolve().parents[2] if str(PROJECT_DEV) not in sys.path: sys.path.insert(0, str(PROJECT_DEV)) from bili_authenticated_extension.constants import CANONICAL_URL, EXTENSION_BUILD, TARGET_BVID # noqa: E402 from bili_authenticated_extension.native_host import ( # noqa: E402 WorkerTask, _valid_control_message, preflight_configuration, ) import bili_authenticated_extension.native_host as native_host_module # noqa: E402 from bili_authenticated_extension.constants import EXPECTED_ORIGIN # noqa: E402 from bili_authenticated_extension.job import create_event, is_event_set, set_event # noqa: E402 from bili_authenticated_extension.protocol import ProtocolError, read_frame, strict_json_loads # noqa: E402 ROOT = PROJECT_DEV / "bili_authenticated_extension" BRIDGE = PROJECT_DEV / "bili_video_download_bridge.py" PROJECT_ROOT = PROJECT_DEV.parents[1] BUILDER_ROOT = PROJECT_ROOT / "dev" / "tmp" / "pyinstaller-6.15.0-offline" BUILDER_PYTHON = BUILDER_ROOT / "venv" / "Scripts" / "python.exe" PYINSTALLER_EXE = BUILDER_ROOT / "venv" / "Scripts" / "pyinstaller.exe" BUILDER_RECEIPT = BUILDER_ROOT / "provision-receipt.md" HISTORICAL_METADATA_TREE_SHA256 = "7C9ABE2387AF3594593BB162C29757B10693F90DD61606272089DB9815C2105C" def digest(path: pathlib.Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest().upper() def metadata_contract(source_manifest: pathlib.Path) -> dict: return json.loads(source_manifest.read_text(encoding="utf-8"))["archive_metadata_contract"] def write_source_approval(path: pathlib.Path, source_manifest: pathlib.Path) -> None: path.write_text(json.dumps({ "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "approval_scope": "controlled-build-source-manifest", "approved_by_role": "dev.reviewer.project", "status": "APPROVED", "source_artifact_manifest_bytes": source_manifest.stat().st_size, "source_artifact_manifest_sha256": digest(source_manifest), }), encoding="utf-8") def write_build_approval( path: pathlib.Path, source_manifest: pathlib.Path, build_manifest: pathlib.Path, host_path: pathlib.Path, ) -> None: path.write_text(json.dumps({ "schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001", "approval_scope": "install-exact-build", "approved_by_role": "dev.reviewer.project", "status": "APPROVED", "source_artifact_manifest_sha256": digest(source_manifest), "build_artifact_manifest_bytes": build_manifest.stat().st_size, "build_artifact_manifest_sha256": digest(build_manifest), "host_executable_bytes": host_path.stat().st_size, "host_executable_sha256": digest(host_path), }), encoding="utf-8") def refresh_source_manifest(source_root: pathlib.Path) -> pathlib.Path: manifest_path = source_root / "source-artifact-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) dependency_manifest = source_root / "dependencies" / "dependency-artifact-manifest.json" manifest["dependency_artifact_manifest_bytes"] = dependency_manifest.stat().st_size manifest["dependency_artifact_manifest_sha256"] = digest(dependency_manifest) for entry in manifest["files"]: product = source_root / entry["path"] entry["bytes"] = product.stat().st_size entry["sha256"] = digest(product) manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return manifest_path def start_message() -> dict: return { "schema": 2, "type": "start", "extension_build": EXTENSION_BUILD, "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "cookie_store_id": "0", "prepare_id": "e" * 32, "page_proof": { "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "task_nonce": "c" * 32, "observed_at_unix_ms": int(time.time() * 1000), "observed_duration_ms": 3_133_950, "video_width": 1920, "video_height": 1080, "ready_state": 4, "eme_present": False, }, "cookies": [{ "name": "SYNTHETIC_JOB_COOKIE", "value": "SYNTHETIC_JOB_VALUE", "domain": ".bilibili.com", "host_only": False, "path": "/", "secure": True, "http_only": True, "same_site": "unspecified", "session": True, "expiration_unix": None, "store_id": "0", "partition_key": None, }], } class NativeHostTests(unittest.TestCase): def test_duplicate_start_does_not_terminate_started_task(self) -> None: class StartedTask: terminal = False prepared = True def __init__(self) -> None: self.terminated = False def start(self, _message: dict) -> None: raise ProtocolError("E_PREPARE") def terminate(self) -> None: self.terminated = True task = StartedTask() self.assertEqual("E_PREPARE", native_host_module._start_prepared_task(task, start_message())) self.assertFalse(task.terminated) def test_worker_control_terminal_cannot_expose_path_or_false_close_marker(self) -> None: complete = { "schema": 1, "type": "terminal", "phase": "COMPLETE", "error_code": None, "formal_filename": f"{TARGET_BVID}.mkv", "mapping_filename": f"{TARGET_BVID}.download.json", "cookie_stream_closed": True, } self.assertTrue(_valid_control_message(complete)) for key, value in ( ("formal_filename", "C:\\secret\\file.mkv"), ("mapping_filename", "../mapping.json"), ("cookie_stream_closed", False), ("error_code", "third-party message"), ): changed = dict(complete) changed[key] = value with self.subTest(key=key): self.assertFalse(_valid_control_message(changed)) def test_broker_checks_origin_before_stdin_and_returns_only_safe_hello(self) -> None: entry = ROOT / "native_host.py" hello = { "schema": 2, "type": "hello", "extension_build": EXTENSION_BUILD, "target": TARGET_BVID, } encoded = json.dumps(hello, separators=(",", ":")).encode("utf-8") framed = struct.pack(" None: ffmpeg = pathlib.Path(subprocess.check_output(["where.exe", "ffmpeg"], text=True).splitlines()[0]) ffprobe = pathlib.Path(subprocess.check_output(["where.exe", "ffprobe"], text=True).splitlines()[0]) with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) destination = root / "destination" destination.mkdir() batch = root / "batch.json" batch.write_text("{}\n", encoding="utf-8") config = root / "config.json" payload = { "schema": 1, "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "ffmpeg": str(ffmpeg), "ffmpeg_sha256": digest(ffmpeg), "ffprobe": str(ffprobe), "ffprobe_sha256": digest(ffprobe), "bridge_python": str(pathlib.Path(sys.executable)), "bridge_python_sha256": digest(pathlib.Path(sys.executable)), "bridge_script": str(BRIDGE), "bridge_script_sha256": digest(BRIDGE), "batch_json": str(batch), "batch_json_sha256": digest(batch), "yt_dlp_executable": str(pathlib.Path(sys.executable)), "yt_dlp_executable_sha256": digest(pathlib.Path(sys.executable)), "destination": str(destination), } config.write_text(json.dumps(payload), encoding="utf-8") self.assertIsNone(preflight_configuration(config)) (destination / f"{TARGET_BVID}.mkv").write_bytes(b"collision") self.assertEqual("E_EXISTS", preflight_configuration(config)) (destination / f"{TARGET_BVID}.mkv").unlink() payload["ffmpeg_sha256"] = "0" * 64 config.write_text(json.dumps(payload), encoding="utf-8") self.assertEqual("E_CONFIG_HASH", preflight_configuration(config)) @unittest.skipUnless(importlib.util.find_spec("yt_dlp"), "frozen yt-dlp environment is required") def test_real_worker_prepares_config_stage_and_stale_cleanup_before_secret(self) -> None: ffmpeg = pathlib.Path(shutil.which("ffmpeg") or "").resolve() ffprobe = pathlib.Path(shutil.which("ffprobe") or "").resolve() with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) local_app_data = root / "local-app-data" local_app_data.mkdir() stage = local_app_data / "project-info" / "bili-auth-ingress" / TARGET_BVID stale = stage / "run-stale" stale.mkdir(parents=True) (stale / "partial.bin").write_bytes(b"stale") destination = root / "destination" destination.mkdir() batch = root / "batch.json" batch.write_text("{}\n", encoding="utf-8") config = root / "config.json" payload = { "schema": 1, "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "ffmpeg": str(ffmpeg), "ffmpeg_sha256": digest(ffmpeg), "ffprobe": str(ffprobe), "ffprobe_sha256": digest(ffprobe), "bridge_python": str(pathlib.Path(sys.executable)), "bridge_python_sha256": digest(pathlib.Path(sys.executable)), "bridge_script": str(BRIDGE), "bridge_script_sha256": digest(BRIDGE), "batch_json": str(batch), "batch_json_sha256": digest(batch), "yt_dlp_executable": str(pathlib.Path(sys.executable)), "yt_dlp_executable_sha256": digest(pathlib.Path(sys.executable)), "destination": str(destination), } config.write_text(json.dumps(payload), encoding="utf-8") task = WorkerTask(config) message = start_message() try: with mock.patch.dict(os.environ, {"LOCALAPPDATA": str(local_app_data)}): task.prepare(message["page_proof"], message["prepare_id"]) self.assertTrue(task.prepared) self.assertEqual("READY", task.phase) self.assertFalse(stale.exists()) leases = list(stage.glob("run-*")) self.assertEqual(1, len(leases)) self.assertEqual(1, len(message["cookies"])) assert task.input_writer is not None task.input_writer.close() task.input_writer = None deadline = time.monotonic() + 10 while time.monotonic() < deadline and not task.terminal: task.poll() time.sleep(0.05) self.assertTrue(task.terminal) self.assertEqual("CANCELED", task.phase) self.assertEqual([], list(destination.iterdir())) self.assertFalse(stage.exists()) finally: task.close() def test_frozen_metadata_and_total_deadlines_are_not_open_ended(self) -> None: task = WorkerTask() task.task_started_at = 100.0 task.phase_started_at = 100.0 task.phase = "CHECKING" self.assertIsNone(task.deadline_error(219.999)) self.assertEqual("E_METADATA_TIMEOUT", task.deadline_error(220.0)) task.phase = "DOWNLOADING" self.assertIsNone(task.deadline_error(7_299.999)) self.assertEqual("E_DOWNLOAD_TIMEOUT", task.deadline_error(7_300.0)) def test_coordinator_cancel_commit_and_timeout_have_unique_terminal(self) -> None: class FakeProcess: def __init__(self) -> None: self.done = False def wait(self, _timeout=0): return self.done def close(self): return None class FakeJob: def __init__(self, process: FakeProcess) -> None: self.process = process def terminate(self): self.process.done = True def close(self): return None nonce = "d" * 32 committed = WorkerTask(ROOT / "config.example.json") committed.cancel_handle = create_event() committed.commit_handle = create_event() committed.secret_started = True committed.task_nonce = nonce set_event(committed.commit_handle) try: self.assertFalse(committed.cancel(nonce)) self.assertFalse(is_event_set(committed.cancel_handle)) self.assertFalse(committed.terminal) finally: committed.close() canceled = WorkerTask(ROOT / "config.example.json") canceled.cancel_handle = create_event() canceled.commit_handle = create_event() canceled.secret_started = True canceled.task_nonce = nonce canceled.process = FakeProcess() canceled.job = FakeJob(canceled.process) with mock.patch.object(native_host_module, "GRACEFUL_CANCEL_SECONDS", 0.01): self.assertTrue(canceled.cancel(nonce)) self.assertTrue(canceled.terminal) self.assertEqual("CANCELED", canceled.phase) self.assertIsNone(canceled.formal_filename) self.assertIsNone(canceled.mapping_filename) timed_out = WorkerTask(ROOT / "config.example.json") timed_out.cancel_handle = create_event() timed_out.commit_handle = create_event() timed_out.secret_started = True timed_out.task_nonce = nonce timed_out.process = FakeProcess() timed_out.job = FakeJob(timed_out.process) timed_out.task_started_at = 1.0 timed_out.phase_started_at = 1.0 timed_out.phase = "DOWNLOADING" with mock.patch.object(native_host_module, "GRACEFUL_CANCEL_SECONDS", 0.01): timed_out.poll() self.assertTrue(timed_out.terminal) self.assertEqual("FAILED", timed_out.phase) self.assertEqual("E_DOWNLOAD_TIMEOUT", timed_out.error_code) self.assertIsNone(timed_out.formal_filename) self.assertIsNone(timed_out.mapping_filename) def test_installer_onefile_receipt_transaction_and_file_registry_provider(self) -> None: script = ROOT / "install_native_host.ps1" with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) config = root / "config.json" destination = root / "destination" destination.mkdir() pinned_python = pathlib.Path(sys.executable) source_file = root / "batch.json" source_file.write_text("{}\n", encoding="ascii") config.write_text(json.dumps({ "schema": 1, "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "ffmpeg": str(pinned_python), "ffmpeg_sha256": digest(pinned_python), "ffprobe": str(pinned_python), "ffprobe_sha256": digest(pinned_python), "bridge_python": str(pinned_python), "bridge_python_sha256": digest(pinned_python), "bridge_script": str(BRIDGE), "bridge_script_sha256": "749FC486B0F42315BD463F11771FE2A7C71CAB53DD9AC2E411CE82E1175DFF13", "batch_json": str(source_file), "batch_json_sha256": digest(source_file), "yt_dlp_executable": str(pinned_python), "yt_dlp_executable_sha256": digest(pinned_python), "destination": str(destination), }), encoding="utf-8") source_manifest = ROOT / "source-artifact-manifest.json" build_root = root / "build" build_root.mkdir() host_path = build_root / "project-info-bili-auth-native-host.exe" shutil.copyfile(sys.executable, host_path) build_manifest = build_root / "build-artifact-manifest.json" build_manifest.write_text(json.dumps({ "schema": 2, "target": TARGET_BVID, "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "extension_build": "project-info-bili-auth-ingress/1.0.0+20260805.v002", "host_build": "project-info-bili-auth-native-host/1.0.0+20260805.v002", "packaging": "pyinstaller-onefile", "pyinstaller_version": "6.15.0", "yt_dlp_version": "2026.7.4", "builder_python_sha256": digest(pathlib.Path(sys.executable)), "pyinstaller_executable_bytes": 108469, "pyinstaller_executable_sha256": "D5DC4427C5E5D417457DAE6FD8B50EF2AFA0A4CE5FDFD5767AB20C5B56555C39", "builder_provision_receipt_bytes": 2027, "builder_provision_receipt_sha256": "B65F4184E8782394F2CC27C47CB8656C942E366FD80E8FA7A548CE5A3367BACF", "build_script_sha256": digest(ROOT / "build_host.ps1"), "source_artifact_manifest_bytes": source_manifest.stat().st_size, "source_artifact_manifest_sha256": digest(source_manifest), "dependency_artifact_manifest_bytes": (ROOT / "dependencies" / "dependency-artifact-manifest.json").stat().st_size, "dependency_artifact_manifest_sha256": digest(ROOT / "dependencies" / "dependency-artifact-manifest.json"), "yt_dlp_wheel_sha256": "F11F2B11D5A8AC4059F9BDF29FA4407DC7C6BB00C5097E95CA22A7A9DB518266", "archive_verification": { "status": "PASS", "method": "PyInstaller 6.15.0 CArchiveReader exact metadata bytes", "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": ["x"], "metadata_tree_sha256": metadata_contract(source_manifest)["canonical_tree_sha256"], }, "files": [{"path": host_path.name, "bytes": host_path.stat().st_size, "sha256": digest(host_path)}], }), encoding="utf-8") trust_root = root / "trust" trust_root.mkdir() source_approval = trust_root / "approved-source.json" build_approval = trust_root / "approved-build.json" write_source_approval(source_approval, source_manifest) write_build_approval(build_approval, source_manifest, build_manifest, host_path) install_root = root / "not-created" registry_root = root / "registry" command = [ "powershell.exe", "-NoProfile", "-File", str(script), "-ObservedExtensionId", "oidmclckpdmpabbfedplkbdplmfcenbb", "-HostExecutable", str(host_path), "-ConfigFile", str(config), "-InstallRoot", str(install_root), "-SourceArtifactManifest", str(source_manifest), "-BuildArtifactManifest", str(build_manifest), "-ApprovedSourceReceipt", str(source_approval), "-ApprovedBuildReceipt", str(build_approval), "-TestFileRegistryProvider", "-TestRegistryRoot", str(registry_root), ] result = subprocess.run(command, check=False, text=True, capture_output=True) self.assertEqual(0, result.returncode, result.stderr) self.assertIn("VALIDATION_PASS_ONLY", result.stdout) self.assertFalse(install_root.exists()) self.assertFalse((root / "registry" / "com.project_info.bili_auth_ingress").exists()) valid_build_manifest_text = build_manifest.read_text(encoding="utf-8") for rejected_hash in (HISTORICAL_METADATA_TREE_SHA256, "0" * 64, "A" * 64): tree_drift = json.loads(valid_build_manifest_text) tree_drift["archive_verification"]["metadata_tree_sha256"] = rejected_hash build_manifest.write_text(json.dumps(tree_drift), encoding="utf-8") write_build_approval(build_approval, source_manifest, build_manifest, host_path) tree_drift_result = subprocess.run(command, check=False, text=True, capture_output=True) self.assertNotEqual(0, tree_drift_result.returncode) self.assertIn("Build artifact receipt identity mismatch", tree_drift_result.stderr) self.assertFalse(install_root.exists()) self.assertFalse((root / "registry" / "com.project_info.bili_auth_ingress").exists()) build_manifest.write_text(valid_build_manifest_text, encoding="utf-8") write_build_approval(build_approval, source_manifest, build_manifest, host_path) wrong = list(command) wrong[wrong.index("oidmclckpdmpabbfedplkbdplmfcenbb")] = "a" * 32 result = subprocess.run(wrong, check=False, text=True, capture_output=True) self.assertNotEqual(0, result.returncode) self.assertFalse(install_root.exists()) extra_source = root / "installer-extra-source" shutil.copytree(ROOT, extra_source) (extra_source / "nested").mkdir() (extra_source / "nested" / "unreviewed.py").write_text("raise SystemExit(99)\n", encoding="utf-8") extra_command = [ str(extra_source / "install_native_host.ps1") if value == str(script) else str(extra_source / "source-artifact-manifest.json") if value == str(source_manifest) else value for value in command ] extra_result = subprocess.run(extra_command, check=False, text=True, capture_output=True) self.assertNotEqual(0, extra_result.returncode) self.assertIn("Actual source tree file set", extra_result.stderr) self.assertFalse(install_root.exists()) arbitrary_manifest = root / "source-artifact-manifest.json" arbitrary_manifest.write_text(source_manifest.read_text(encoding="utf-8"), encoding="utf-8") arbitrary = list(command) arbitrary[arbitrary.index(str(source_manifest))] = str(arbitrary_manifest) result = subprocess.run(arbitrary, check=False, text=True, capture_output=True) self.assertNotEqual(0, result.returncode) install_command = [ *command, "-Install", ] result = subprocess.run(install_command, check=False, text=True, capture_output=True) self.assertEqual(0, result.returncode, result.stderr) self.assertTrue((install_root / host_path.name).is_file()) registry_key = registry_root / "com.project_info.bili_auth_ingress" self.assertEqual( str(install_root / "native-host-manifest.json"), (registry_key / "default.value").read_text(encoding="utf-8"), ) repeated = subprocess.run(install_command, check=False, text=True, capture_output=True) self.assertNotEqual(0, repeated.returncode) for point in ( "after-root", "after-payload", "after-config", "after-manifest", "after-registry-key", "after-registry-value", ): case_install = root / f"install-{point}" case_registry = root / f"registry-{point}" case_command = list(command) case_command[case_command.index(str(install_root))] = str(case_install) case_command[case_command.index(str(registry_root))] = str(case_registry) case_command.extend([ "-Install", "-InjectFailure", point, ]) failed = subprocess.run(case_command, check=False, text=True, capture_output=True) self.assertNotEqual(0, failed.returncode, point) self.assertFalse(case_install.exists(), point) self.assertFalse((case_registry / "com.project_info.bili_auth_ingress").exists(), point) host_path.write_bytes(host_path.read_bytes() + b"tamper") tampered = subprocess.run(command, check=False, text=True, capture_output=True) self.assertNotEqual(0, tampered.returncode) altered_source = root / "altered-source" shutil.copytree(ROOT, altered_source) with (altered_source / "worker.py").open("ab") as target: target.write(b"\n# synthetic source drift\n") altered_manifest = refresh_source_manifest(altered_source) altered_build = root / "altered-build" altered_build.mkdir() altered_host = altered_build / host_path.name shutil.copyfile(sys.executable, altered_host) altered_build_manifest = altered_build / build_manifest.name altered_build_manifest.write_text(json.dumps({ "schema": 2, "target": TARGET_BVID, "extension_id": "oidmclckpdmpabbfedplkbdplmfcenbb", "extension_build": "project-info-bili-auth-ingress/1.0.0+20260805.v002", "host_build": "project-info-bili-auth-native-host/1.0.0+20260805.v002", "packaging": "pyinstaller-onefile", "pyinstaller_version": "6.15.0", "yt_dlp_version": "2026.7.4", "builder_python_sha256": digest(pathlib.Path(sys.executable)), "pyinstaller_executable_bytes": 108469, "pyinstaller_executable_sha256": "D5DC4427C5E5D417457DAE6FD8B50EF2AFA0A4CE5FDFD5767AB20C5B56555C39", "builder_provision_receipt_bytes": 2027, "builder_provision_receipt_sha256": "B65F4184E8782394F2CC27C47CB8656C942E366FD80E8FA7A548CE5A3367BACF", "build_script_sha256": digest(altered_source / "build_host.ps1"), "source_artifact_manifest_bytes": altered_manifest.stat().st_size, "source_artifact_manifest_sha256": digest(altered_manifest), "dependency_artifact_manifest_bytes": (altered_source / "dependencies" / "dependency-artifact-manifest.json").stat().st_size, "dependency_artifact_manifest_sha256": digest(altered_source / "dependencies" / "dependency-artifact-manifest.json"), "yt_dlp_wheel_sha256": "F11F2B11D5A8AC4059F9BDF29FA4407DC7C6BB00C5097E95CA22A7A9DB518266", "archive_verification": { "status": "PASS", "method": "PyInstaller 6.15.0 CArchiveReader exact metadata bytes", "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": ["x"], "metadata_tree_sha256": metadata_contract(altered_manifest)["canonical_tree_sha256"], }, "files": [{"path": altered_host.name, "bytes": altered_host.stat().st_size, "sha256": digest(altered_host)}], }), encoding="utf-8") altered_build_approval = trust_root / "altered-build-approved.json" write_build_approval(altered_build_approval, altered_manifest, altered_build_manifest, altered_host) altered_install = root / "altered-install" altered_registry = root / "altered-registry" altered_command = list(command) replacements = { str(script): str(altered_source / "install_native_host.ps1"), str(host_path): str(altered_host), str(source_manifest): str(altered_manifest), str(build_manifest): str(altered_build_manifest), str(build_approval): str(altered_build_approval), str(install_root): str(altered_install), str(registry_root): str(altered_registry), } altered_command = [replacements.get(value, value) for value in altered_command] altered_command.append("-Install") altered_result = subprocess.run(altered_command, check=False, text=True, capture_output=True) self.assertNotEqual(0, altered_result.returncode) self.assertIn("approved source receipt", altered_result.stderr.lower()) self.assertFalse(altered_install.exists()) self.assertFalse((altered_registry / "com.project_info.bili_auth_ingress").exists()) def test_build_script_rejects_unpinned_or_missing_builder_before_output(self) -> None: script = ROOT / "build_host.ps1" source = script.read_text(encoding="utf-8") self.assertIn("--onefile", source) self.assertNotIn("--onedir", source) with tempfile.TemporaryDirectory() as temporary: temporary_root = pathlib.Path(temporary) output_root = temporary_root / "not-created" trust_root = temporary_root / "trust" trust_root.mkdir() source_approval = trust_root / "approved-source.json" write_source_approval(source_approval, ROOT / "source-artifact-manifest.json") result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(script), "-Python", sys.executable, "-PyInstallerExecutable", sys.executable, "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(output_root), "-ApprovedSourceReceipt", str(source_approval), "-SourceArtifactManifest", str(ROOT / "source-artifact-manifest.json"), ], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.assertNotEqual(0, result.returncode) self.assertIn("Pinned builder executables", result.stderr) self.assertFalse(output_root.exists()) copied_source = temporary_root / "copied-source" shutil.copytree(ROOT, copied_source) with (copied_source / "worker.py").open("ab") as target: target.write(b"\n# tampered\n") tampered_output = temporary_root / "tampered-source-output" tampered_build = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(copied_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(tampered_output), "-ApprovedSourceReceipt", str(source_approval), "-SourceArtifactManifest", str(copied_source / "source-artifact-manifest.json"), ], check=False, text=True, capture_output=True, ) self.assertNotEqual(0, tampered_build.returncode) self.assertIn("Source artifact hash mismatch", tampered_build.stderr) self.assertFalse(tampered_output.exists()) for case_name, extra_path in ( ("extra-root", pathlib.Path("yt_dlp.py")), ("extra-nested", pathlib.Path("nested") / "unreviewed.py"), ): extra_source = temporary_root / case_name shutil.copytree(ROOT, extra_source) (extra_source / extra_path).parent.mkdir(parents=True, exist_ok=True) (extra_source / extra_path).write_text("raise SystemExit(91)\n", encoding="utf-8") extra_approval = trust_root / f"{case_name}-approved.json" write_source_approval(extra_approval, extra_source / "source-artifact-manifest.json") extra_output = temporary_root / f"{case_name}-output" extra_result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(extra_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(extra_output), "-ApprovedSourceReceipt", str(extra_approval), "-SourceArtifactManifest", str(extra_source / "source-artifact-manifest.json"), "-ValidateOnly", ], check=False, text=True, capture_output=True, ) self.assertNotEqual(0, extra_result.returncode, case_name) self.assertIn("Actual source tree file set", extra_result.stderr, case_name) self.assertFalse(extra_output.exists(), case_name) reparse_source = temporary_root / "reparse-source" shutil.copytree(ROOT, reparse_source) reparse_target = temporary_root / "reparse-target" reparse_target.mkdir() (reparse_target / "unreviewed.py").write_text("raise SystemExit(92)\n", encoding="utf-8") junction = reparse_source / "linked-source" junction_environment = dict( os.environ, TEST_JUNCTION_LINK=str(junction), TEST_JUNCTION_TARGET=str(reparse_target), ) junction_result = subprocess.run( [ "powershell.exe", "-NoProfile", "-Command", "New-Item -ItemType Junction -Path $env:TEST_JUNCTION_LINK -Target $env:TEST_JUNCTION_TARGET | Out-Null", ], check=False, text=True, capture_output=True, env=junction_environment, ) self.assertEqual(0, junction_result.returncode, junction_result.stderr) reparse_approval = trust_root / "reparse-source-approved.json" write_source_approval(reparse_approval, reparse_source / "source-artifact-manifest.json") reparse_output = temporary_root / "reparse-output" reparse_result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(reparse_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(reparse_output), "-ApprovedSourceReceipt", str(reparse_approval), "-SourceArtifactManifest", str(reparse_source / "source-artifact-manifest.json"), "-ValidateOnly", ], check=False, text=True, capture_output=True, ) self.assertNotEqual(0, reparse_result.returncode) self.assertIn("reparse path", reparse_result.stderr) self.assertFalse(reparse_output.exists()) fake_site = temporary_root / "fake-site" package = fake_site / "PyInstaller" metadata = fake_site / "PyInstaller-6.15.0.dist-info" package.mkdir(parents=True) metadata.mkdir() (metadata / "METADATA").write_text( "Metadata-Version: 2.1\nName: PyInstaller\nVersion: 6.15.0\n", encoding="utf-8", ) (package / "__init__.py").write_text("", encoding="utf-8") (package / "__main__.py").write_text( "import os, pathlib, sys\n" "args=sys.argv[1:]\n" "if '--onefile' not in args or '--onedir' in args: raise SystemExit(8)\n" "if args[args.index('--collect-all')+1] != 'yt_dlp': raise SystemExit(10)\n" "if args[args.index('--copy-metadata')+1] != 'yt-dlp': raise SystemExit(11)\n" "if '--paths' not in args: raise SystemExit(12)\n" "def value(flag): return args[args.index(flag)+1]\n" "dist=pathlib.Path(value('--distpath')); work=pathlib.Path(value('--workpath')); spec=pathlib.Path(value('--specpath'))\n" "dist.mkdir(parents=True,exist_ok=True); work.mkdir(parents=True,exist_ok=True); spec.mkdir(parents=True,exist_ok=True)\n" "(dist/(value('--name')+'.exe')).write_bytes(b'synthetic-onefile')\n" "raise SystemExit(9 if os.environ.get('FAKE_PYINSTALLER_FAIL') else 0)\n", encoding="utf-8", ) archive_package = package / "archive" archive_package.mkdir(parents=True) (archive_package / "__init__.py").write_text("", encoding="utf-8") (archive_package / "readers.py").write_text( "import os, pathlib\n" "class Embedded:\n" " def __init__(self):\n" " names={'bili_authenticated_extension.worker','yt_dlp','yt_dlp.downloader','yt_dlp.globals','yt_dlp.plugins','yt_dlp.version'}\n" " if os.environ.get('FAKE_ARCHIVE_OMIT_YTDLP'): names.discard('yt_dlp.downloader')\n" " self.toc={name:None for name in names}\n" "class CArchiveReader:\n" " def __init__(self, executable):\n" " root=pathlib.Path(executable).parent/'.deps'/'yt_dlp-2026.7.4.dist-info'\n" " payloads={p.relative_to(root).as_posix():p.read_bytes() for p in root.rglob('*') if p.is_file()}\n" " if os.environ.get('FAKE_ARCHIVE_OMIT_METADATA'): payloads={}\n" " if os.environ.get('FAKE_ARCHIVE_INCOMPLETE_METADATA'): payloads.pop('WHEEL',None)\n" " if os.environ.get('FAKE_ARCHIVE_WRONG_VERSION') and 'METADATA' in payloads:\n" " payloads['METADATA']=payloads['METADATA'].replace(b'\\nVersion: 2026.7.4\\n',b'\\nVersion: 2026.7.3\\n',1)\n" " if os.environ.get('FAKE_ARCHIVE_WRONG_NAME') and 'METADATA' in payloads:\n" " payloads['METADATA']=payloads['METADATA'].replace(b'\\nName: yt-dlp\\n',b'\\nName: unrelated\\n',1)\n" " if os.environ.get('FAKE_ARCHIVE_WRONG_BYTES') and 'entry_points.txt' in payloads:\n" " payloads['entry_points.txt']+=b'\\n'\n" " archive_root='yt_dlp-2026.7.4.dist-info'\n" " if os.environ.get('FAKE_ARCHIVE_UNEXPECTED_PATH'): archive_root='nested/'+archive_root\n" " if os.environ.get('FAKE_ARCHIVE_SUBSTITUTE'): archive_root='unrelated-1.0.dist-info'\n" " self.payloads={}\n" " self.toc={'PYZ.pyz':(0,0,0,0,'z')}\n" " metadata_type=os.environ.get('FAKE_ARCHIVE_METADATA_TYPE','x')\n" " for relative,payload in payloads.items():\n" " raw=(archive_root+'/'+relative).replace('/','\\\\')\n" " self.payloads[raw]=payload; self.toc[raw]=(0,0,0,0,metadata_type)\n" " if os.environ.get('FAKE_ARCHIVE_DUPLICATE_METADATA') and self.payloads:\n" " raw=next(iter(self.payloads)); duplicate=raw.replace('\\\\','/')\n" " self.payloads[duplicate]=self.payloads[raw]; self.toc[duplicate]=(0,0,0,0,'x')\n" " def extract(self, name): return self.payloads[name]\n" " def open_embedded_archive(self, name): return Embedded()\n", encoding="utf-8", ) injection_marker = temporary_root / "pythonpath-injection-executed.txt" (package / "__init__.py").write_text( f"import pathlib\npathlib.Path({str(injection_marker)!r}).write_text('PyInstaller imported')\n", encoding="utf-8", ) fake_pip = fake_site / "pip" fake_pip.mkdir() (fake_pip / "__init__.py").write_text("", encoding="utf-8") (fake_pip / "__main__.py").write_text( f"import pathlib\npathlib.Path({str(injection_marker)!r}).write_text('pip executed')\nraise SystemExit(95)\n", encoding="utf-8", ) environment = os.environ.copy() environment["PYTHONPATH"] = str(fake_site) production_validate_output = temporary_root / "production-validate-output" production_validate_command = [ "powershell.exe", "-NoProfile", "-File", str(script), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(production_validate_output), "-ApprovedSourceReceipt", str(source_approval), "-SourceArtifactManifest", str(ROOT / "source-artifact-manifest.json"), "-ValidateOnly", ] validated = subprocess.run( production_validate_command, check=False, text=True, capture_output=True, env=environment, ) self.assertEqual(0, validated.returncode, validated.stderr) self.assertIn("VALIDATION_PASS_ONLY", validated.stdout) self.assertFalse(production_validate_output.exists()) self.assertFalse(injection_marker.exists()) shutil.rmtree(fake_pip) fake_pyinstaller_exe = temporary_root / "fake-pyinstaller.exe" fake_pyinstaller_exe.write_bytes(b"synthetic-pyinstaller-executable") synthetic_builder_receipt = temporary_root / "synthetic-builder-provision.md" synthetic_builder_receipt.write_text( "# Synthetic test-only builder provision\n\n" "request_handoff: HANDOFF-INFOADMIN-MGADMIN-BILI-PYINSTALLER-6-15-0-OFFLINE-ENV-PROVISION-20260806-001\n" "scope: project-info 私有、隔离、可离线复用\n" f"environment_python: `{BUILDER_PYTHON}`\n" f"environment_python_bytes_sha256: `{BUILDER_PYTHON.stat().st_size}/{digest(BUILDER_PYTHON).lower()}`\n" f"pyinstaller_executable: `{fake_pyinstaller_exe}`\n" f"pyinstaller_executable_bytes_sha256: `{fake_pyinstaller_exe.stat().st_size}/{digest(fake_pyinstaller_exe).lower()}`\n", encoding="utf-8", ) seam_source = temporary_root / "test-only-builder-seam-source" shutil.copytree(ROOT, seam_source) seam_script = seam_source / "build_host.ps1" seam_text = seam_script.read_text(encoding="utf-8") replacements = { "$expectedBuilderReceiptBytes = 2027": f"$expectedBuilderReceiptBytes = {synthetic_builder_receipt.stat().st_size}", "$expectedBuilderReceiptSha256 = 'B65F4184E8782394F2CC27C47CB8656C942E366FD80E8FA7A548CE5A3367BACF'": f"$expectedBuilderReceiptSha256 = '{digest(synthetic_builder_receipt)}'", "$expectedBuilderPythonBytes = 262144": f"$expectedBuilderPythonBytes = {BUILDER_PYTHON.stat().st_size}", "$expectedBuilderPythonSha256 = '5912D0884B23C0343983A864C6064242391E2265536F50B88624857E353882C9'": f"$expectedBuilderPythonSha256 = '{digest(BUILDER_PYTHON)}'", "$expectedPyInstallerBytes = 108469": f"$expectedPyInstallerBytes = {fake_pyinstaller_exe.stat().st_size}", "$expectedPyInstallerSha256 = 'D5DC4427C5E5D417457DAE6FD8B50EF2AFA0A4CE5FDFD5767AB20C5B56555C39'": f"$expectedPyInstallerSha256 = '{digest(fake_pyinstaller_exe)}'", } for before, after in replacements.items(): self.assertIn(before, seam_text) seam_text = seam_text.replace(before, after) seam_text = seam_text.replace("-I -S -B", "-S -B").replace("-I -B", "-B") seam_marker = " $pyInstallerVersion = & $resolvedPython -B" self.assertIn(seam_marker, seam_text) fake_site_ps = str(fake_site).replace("'", "''") seam_text = seam_text.replace( seam_marker, f" $env:PYTHONPATH = '{fake_site_ps}'\n{seam_marker}", 1, ) seam_script.write_text(seam_text, encoding="utf-8") seam_manifest = refresh_source_manifest(seam_source) seam_approval = trust_root / "test-only-seam-source-approved.json" write_source_approval(seam_approval, seam_manifest) built = temporary_root / "built" build_command = [ "powershell.exe", "-NoProfile", "-File", str(seam_script), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(fake_pyinstaller_exe), "-ApprovedBuilderReceipt", str(synthetic_builder_receipt), "-OutputRoot", str(built), "-ApprovedSourceReceipt", str(seam_approval), "-SourceArtifactManifest", str(seam_manifest), ] result = subprocess.run(build_command, check=False, text=True, capture_output=True, env=environment) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual( {"project-info-bili-auth-native-host.exe", "build-artifact-manifest.json"}, {item.name for item in built.iterdir()}, ) receipt = json.loads((built / "build-artifact-manifest.json").read_text(encoding="utf-8")) self.assertEqual("pyinstaller-onefile", receipt["packaging"]) self.assertEqual("2026.7.4", receipt["yt_dlp_version"]) self.assertEqual("PASS", receipt["archive_verification"]["status"]) self.assertEqual(["x"], receipt["archive_verification"]["metadata_type_codes"]) self.assertEqual( digest(seam_source / "dependencies" / "dependency-artifact-manifest.json"), receipt["dependency_artifact_manifest_sha256"], ) self.assertEqual(1, len(receipt["files"])) self.assertEqual("project-info-bili-auth-native-host.exe", receipt["files"][0]["path"]) promoted_output = temporary_root / "promoted-b-build" promoted_command = list(build_command) promoted_command[promoted_command.index(str(built))] = str(promoted_output) promoted_environment = dict(environment, FAKE_ARCHIVE_METADATA_TYPE="b") promoted = subprocess.run( promoted_command, check=False, text=True, capture_output=True, env=promoted_environment, ) self.assertEqual(0, promoted.returncode, promoted.stderr) promoted_receipt = json.loads( (promoted_output / "build-artifact-manifest.json").read_text(encoding="utf-8") ) self.assertEqual(["b"], promoted_receipt["archive_verification"]["metadata_type_codes"]) for rejected_type in ("Z", "a", "d", "l", "m", "n", "o", "s", "z"): rejected_output = temporary_root / f"rejected-type-{ord(rejected_type)}" rejected_command = list(build_command) rejected_command[rejected_command.index(str(built))] = str(rejected_output) rejected_environment = dict(environment, FAKE_ARCHIVE_METADATA_TYPE=rejected_type) rejected = subprocess.run( rejected_command, check=False, text=True, capture_output=True, env=rejected_environment, ) self.assertNotEqual(0, rejected.returncode, rejected_type) self.assertIn("E_ARCHIVE_METADATA_TYPE", rejected.stderr, rejected_type) self.assertFalse(rejected_output.exists(), rejected_type) unbound_b_output = temporary_root / "unbound-b-metadata" unbound_b_command = list(build_command) unbound_b_command[unbound_b_command.index(str(built))] = str(unbound_b_output) unbound_b_environment = dict( environment, FAKE_ARCHIVE_METADATA_TYPE="b", FAKE_ARCHIVE_UNEXPECTED_PATH="1", ) unbound_b = subprocess.run( unbound_b_command, check=False, text=True, capture_output=True, env=unbound_b_environment, ) self.assertNotEqual(0, unbound_b.returncode) self.assertIn("E_ARCHIVE_METADATA_DUPLICATE_OR_PATH", unbound_b.stderr) self.assertFalse(unbound_b_output.exists()) resigned_source = temporary_root / "resigned-source" shutil.copytree(ROOT, resigned_source) with (resigned_source / "worker.py").open("ab") as target: target.write(b"\n# synthetic source drift\n") resigned_manifest = refresh_source_manifest(resigned_source) resigned_output = temporary_root / "resigned-output" resigned_command = [ "powershell.exe", "-NoProfile", "-File", str(resigned_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(resigned_output), "-ApprovedSourceReceipt", str(source_approval), "-SourceArtifactManifest", str(resigned_manifest), ] resigned = subprocess.run(resigned_command, check=False, text=True, capture_output=True, env=environment) self.assertNotEqual(0, resigned.returncode) self.assertIn("approved source receipt", resigned.stderr.lower()) self.assertFalse(resigned_output.exists()) failed_output = temporary_root / "failed-build" failed_command = list(build_command) failed_command[failed_command.index(str(built))] = str(failed_output) failed_environment = dict(environment, FAKE_PYINSTALLER_FAIL="1") result = subprocess.run(failed_command, check=False, text=True, capture_output=True, env=failed_environment) self.assertNotEqual(0, result.returncode) self.assertFalse(failed_output.exists()) for variable, expected_error in ( ("FAKE_ARCHIVE_OMIT_YTDLP", "E_ARCHIVE_MODULE_MISSING"), ("FAKE_ARCHIVE_OMIT_METADATA", "E_ARCHIVE_METADATA_MISSING"), ("FAKE_ARCHIVE_DUPLICATE_METADATA", "E_ARCHIVE_DUPLICATE_PATH"), ("FAKE_ARCHIVE_WRONG_VERSION", "E_ARCHIVE_METADATA_IDENTITY"), ("FAKE_ARCHIVE_WRONG_NAME", "E_ARCHIVE_METADATA_IDENTITY"), ("FAKE_ARCHIVE_WRONG_BYTES", "E_ARCHIVE_METADATA_BYTES"), ("FAKE_ARCHIVE_INCOMPLETE_METADATA", "E_ARCHIVE_METADATA_FILE_SET"), ("FAKE_ARCHIVE_UNEXPECTED_PATH", "E_ARCHIVE_METADATA_DUPLICATE_OR_PATH"), ("FAKE_ARCHIVE_SUBSTITUTE", "E_ARCHIVE_METADATA_DUPLICATE_OR_PATH"), ): archive_output = temporary_root / variable.lower() archive_command = list(build_command) archive_command[archive_command.index(str(built))] = str(archive_output) archive_environment = dict(environment, **{variable: "1"}) archived = subprocess.run( archive_command, check=False, text=True, capture_output=True, env=archive_environment, ) self.assertNotEqual(0, archived.returncode) self.assertIn(expected_error, archived.stderr) self.assertFalse(archive_output.exists()) incomplete_source = temporary_root / "incomplete-dependency-source" shutil.copytree(ROOT, incomplete_source) missing_wheel = incomplete_source / "dependencies" / "yt_dlp-2026.7.4-py3-none-any.whl" missing_wheel.unlink() incomplete_manifest_path = incomplete_source / "source-artifact-manifest.json" incomplete_manifest = json.loads(incomplete_manifest_path.read_text(encoding="utf-8")) incomplete_manifest["files"] = [ item for item in incomplete_manifest["files"] if item["path"] != "dependencies/yt_dlp-2026.7.4-py3-none-any.whl" ] incomplete_manifest_path.write_text( json.dumps(incomplete_manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) incomplete_approval = trust_root / "incomplete-source-approved.json" write_source_approval(incomplete_approval, incomplete_manifest_path) incomplete_output = temporary_root / "incomplete-output" incomplete_result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(incomplete_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(incomplete_output), "-ApprovedSourceReceipt", str(incomplete_approval), "-SourceArtifactManifest", str(incomplete_manifest_path), ], check=False, text=True, capture_output=True, env=environment, ) self.assertNotEqual(0, incomplete_result.returncode) self.assertIn("Actual source tree file set", incomplete_result.stderr) self.assertFalse(incomplete_output.exists()) tampered_dependency_source = temporary_root / "tampered-dependency-source" shutil.copytree(ROOT, tampered_dependency_source) tampered_wheel = tampered_dependency_source / "dependencies" / "yt_dlp-2026.7.4-py3-none-any.whl" tampered_wheel.write_bytes(tampered_wheel.read_bytes() + b"tamper") tampered_dependency_manifest = refresh_source_manifest(tampered_dependency_source) tampered_dependency_approval = trust_root / "tampered-dependency-approved.json" write_source_approval(tampered_dependency_approval, tampered_dependency_manifest) tampered_dependency_output = temporary_root / "tampered-dependency-output" tampered_dependency_result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(tampered_dependency_source / "build_host.ps1"), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(tampered_dependency_output), "-ApprovedSourceReceipt", str(tampered_dependency_approval), "-SourceArtifactManifest", str(tampered_dependency_manifest), ], check=False, text=True, capture_output=True, env=environment, ) self.assertNotEqual(0, tampered_dependency_result.returncode) self.assertIn("Dependency wheelhouse file set or hash mismatch", tampered_dependency_result.stderr) self.assertFalse(tampered_dependency_output.exists()) def test_real_carchive_writer_accepts_x_and_windows_promoted_b_metadata(self) -> None: build_source = (ROOT / "build_host.ps1").read_text(encoding="utf-8") start_marker = "$archiveProbeScript = @'\n" end_marker = "\n'@\n $archiveProbeJson =" self.assertIn(start_marker, build_source) self.assertIn(end_marker, build_source) probe = build_source.split(start_marker, 1)[1].split(end_marker, 1)[0] wheel_path = ROOT / "dependencies" / "yt_dlp-2026.7.4-py3-none-any.whl" with tempfile.TemporaryDirectory() as temporary: root = pathlib.Path(temporary) dependency_root = root / ".deps" metadata_root = dependency_root / "yt_dlp-2026.7.4.dist-info" installed = subprocess.run( [ str(BUILDER_PYTHON), "-I", "-B", "-m", "pip", "install", "--no-index", "--find-links", str(wheel_path.parent), "--no-deps", "--no-compile", "--target", str(dependency_root), "yt-dlp==2026.7.4", ], check=False, text=True, capture_output=True, env=dict( os.environ, SOURCE_DATE_EPOCH=str(metadata_contract(ROOT / "source-artifact-manifest.json")["source_date_epoch"]), ), ) self.assertEqual(0, installed.returncode, installed.stderr) self.assertTrue(metadata_root.is_dir()) self.assertTrue(os.access(metadata_root / "METADATA", os.X_OK)) writer = ( "import pathlib,sys\n" "from PyInstaller.archive.writers import CArchiveWriter\n" "out=pathlib.Path(sys.argv[1]); metadata=pathlib.Path(sys.argv[2]); kind=sys.argv[3]\n" "dummy=out.with_suffix('.payload'); dummy.write_bytes(b'anchor')\n" "modules={'bili_authenticated_extension.worker','yt_dlp','yt_dlp.downloader','yt_dlp.globals','yt_dlp.plugins','yt_dlp.version'}\n" "entries=[(name,str(dummy),True,'x') for name in sorted(modules)]\n" "entries += [('yt_dlp-2026.7.4.dist-info/'+p.relative_to(metadata).as_posix(),str(p),True,kind) for p in sorted(metadata.rglob('*')) if p.is_file()]\n" "CArchiveWriter(str(out),entries,'python311.dll')\n" ) for metadata_type in ("x", "b"): archive_path = root / f"metadata-{metadata_type}.pkg" written = subprocess.run( [ str(BUILDER_PYTHON), "-I", "-B", "-c", writer, str(archive_path), str(metadata_root), metadata_type, ], check=False, text=True, capture_output=True, ) self.assertEqual(0, written.returncode, written.stderr) checked = subprocess.run( [ str(BUILDER_PYTHON), "-I", "-B", "-c", probe, str(archive_path), str(dependency_root), str(ROOT / "source-artifact-manifest.json"), ], check=False, text=True, capture_output=True, ) self.assertEqual(0, checked.returncode, checked.stderr) result = json.loads(checked.stdout) self.assertEqual("PASS", result["status"], result) self.assertEqual([metadata_type], result["metadata_type_codes"], result) def test_dependency_tree_metadata_discovery_is_exact_and_injection_safe(self) -> None: with tempfile.TemporaryDirectory() as temporary: temporary_root = pathlib.Path(temporary) trust_root = temporary_root / "trust" trust_root.mkdir() def run_modified(name: str, modify, environment=None): source = temporary_root / f"source-{name}" shutil.copytree(ROOT, source) script = source / "build_host.ps1" text = script.read_text(encoding="utf-8") updated = modify(text) self.assertNotEqual(text, updated) script.write_text(updated, encoding="utf-8") manifest = refresh_source_manifest(source) approval = trust_root / f"approved-{name}.json" write_source_approval(approval, manifest) output = temporary_root / f"output-{name}" result = subprocess.run( [ "powershell.exe", "-NoProfile", "-File", str(script), "-Python", str(BUILDER_PYTHON), "-PyInstallerExecutable", str(PYINSTALLER_EXE), "-ApprovedBuilderReceipt", str(BUILDER_RECEIPT), "-OutputRoot", str(output), "-ApprovedSourceReceipt", str(approval), "-SourceArtifactManifest", str(manifest), ], check=False, text=True, capture_output=True, env=environment, ) self.assertNotEqual(0, result.returncode) self.assertFalse(output.exists()) self.assertFalse((output / "build-artifact-manifest.json").exists()) return result post_install_validation_anchor = ( " $dependencyRootItem = Get-Item -LiteralPath $dependencyRoot" ) def inject_after_install(statement: str): def modifier(text: str) -> str: self.assertIn(post_install_validation_anchor, text) return text.replace( post_install_validation_anchor, f" {statement}\n{post_install_validation_anchor}", 1, ) return modifier failures = ( ( "missing-metadata", inject_after_install( "Remove-Item -LiteralPath (Join-Path $dependencyRoot 'yt_dlp-2026.7.4.dist-info\\METADATA') -Force" ), "Temporary dependency tree does not exactly match", ), ( "tampered-package", inject_after_install( "[IO.File]::AppendAllText((Join-Path $dependencyRoot 'yt_dlp\\version.py'), '# test tamper')" ), "Temporary dependency tree does not exactly match", ), ( "extra-distribution", inject_after_install( "$extra = Join-Path $dependencyRoot 'unexpected-1.0.dist-info'; " "[IO.Directory]::CreateDirectory($extra) | Out-Null; " "[IO.File]::WriteAllText((Join-Path $extra 'METADATA'), " "'Metadata-Version: 2.1`nName: unexpected`nVersion: 1.0`n'); " "[IO.File]::WriteAllText((Join-Path $extra 'RECORD'), '')" ), "Temporary dependency tree does not exactly match", ), ) for name, modifier, expected_error in failures: with self.subTest(name=name): failed = run_modified(name, modifier) self.assertIn(expected_error, failed.stderr) dependency_import_marker = temporary_root / "tampered-dependency-import-executed.txt" dependency_import_marker_path = dependency_import_marker.as_posix() tampered_import = run_modified( "tampered-first-import-module", inject_after_install( "$payload = \"`nimport pathlib`npathlib.Path(r'" + dependency_import_marker_path + "').write_text('executed', encoding='utf-8')`n\"; " "[IO.File]::AppendAllText((Join-Path $dependencyRoot 'yt_dlp\\__init__.py'), $payload)" ), ) self.assertIn("Temporary dependency tree does not exactly match", tampered_import.stderr) self.assertFalse(dependency_import_marker.exists()) tree_call = ( "$dependencyTreeJson = & $resolvedPython -I -S -B -c " "$dependencyTreeProbeScript $dependencyRoot $resolvedOutput $wheelPath" ) def wrong_owner(text: str) -> str: self.assertIn(tree_call, text) return text.replace( tree_call, tree_call.replace("$resolvedOutput", "$dependencyWheelhouse"), 1, ) wrong = run_modified("wrong-owner", wrong_owner) self.assertIn("Temporary dependency tree does not exactly match", wrong.stderr) fake_site = temporary_root / "fake-site" fake_pyinstaller = fake_site / "PyInstaller" fake_pip = fake_site / "pip" fake_pyinstaller.mkdir(parents=True) fake_pip.mkdir() marker = temporary_root / "inherited-pythonpath-executed.txt" marker_code = ( "import pathlib\n" f"pathlib.Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n" ) (fake_pyinstaller / "__init__.py").write_text(marker_code, encoding="utf-8") (fake_pyinstaller / "__main__.py").write_text(marker_code, encoding="utf-8") (fake_pip / "__init__.py").write_text(marker_code, encoding="utf-8") (fake_pip / "__main__.py").write_text(marker_code, encoding="utf-8") environment = os.environ.copy() environment["PYTHONPATH"] = str(fake_site) launch_anchor = "$pyInstallerLaunchScript = @'" def stop_after_real_metadata(text: str) -> str: self.assertIn(launch_anchor, text) return text.replace( launch_anchor, " throw 'TEST_REAL_METADATA_DISCOVERY_PASS'\n" + launch_anchor, 1, ) discovered = run_modified( "real-metadata-clean-environment", stop_after_real_metadata, environment, ) self.assertIn("TEST_REAL_METADATA_DISCOVERY_PASS", discovered.stderr) self.assertFalse(marker.exists()) if __name__ == "__main__": unittest.main()