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