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("<I", len(encoded)) + encoded
|
wrong = subprocess.run(
|
[sys.executable, str(entry), "chrome-extension://wrong/"],
|
input=framed,
|
stdout=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
timeout=10,
|
)
|
self.assertEqual(2, wrong.returncode)
|
self.assertEqual(b"", wrong.stdout)
|
self.assertEqual(b"", wrong.stderr)
|
|
accepted = subprocess.run(
|
[sys.executable, str(entry), EXPECTED_ORIGIN, "--parent-window=17"],
|
input=framed,
|
stdout=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
timeout=10,
|
)
|
self.assertEqual(0, accepted.returncode)
|
stream = io.BytesIO(accepted.stdout)
|
payload = read_frame(stream)
|
self.assertIsNotNone(payload)
|
response = strict_json_loads(payload)
|
self.assertEqual("hello", response["type"])
|
self.assertEqual("FAILED", response["phase"])
|
self.assertEqual("E_CONFIG", response["error_code"])
|
self.assertIsNone(read_frame(stream))
|
self.assertEqual(b"", accepted.stderr)
|
|
def test_preflight_hashes_and_collision_run_before_secret(self) -> 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()
|
