#!/usr/bin/env python3 """Offline validator for the local-unpacked Bilibili extension projection.""" from __future__ import annotations import argparse import base64 import ctypes from ctypes import wintypes from dataclasses import asdict, dataclass import hashlib import json import os from pathlib import Path, PurePosixPath import re import stat import sys from typing import Callable, Iterable SCHEMA = 1 PROJECT_ID = "project-info" TASK_ID = "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001" SOURCE_ROOT_REL = PurePosixPath("dev/project-dev/bili_authenticated_extension") PROJECTION_ROOT_REL = PurePosixPath("dev/project-dev/bili_authenticated_extension_unpacked") CONTRACT_REL = PurePosixPath("dev/project-dev/bili_authenticated_extension_unpacked_contract.json") SOURCE_ARTIFACT_MANIFEST_REL = SOURCE_ROOT_REL / "source-artifact-manifest.json" PROJECT_MARKER_REL = PurePosixPath("mbx.project.yaml") EXPECTED_FILES = ( "background.js", "manifest.json", "sidepanel.css", "sidepanel.html", "sidepanel.js", ) EXPECTED_EXTENSION_ID = "oidmclckpdmpabbfedplkbdplmfcenbb" EXPECTED_PUBLIC_DER_SHA256 = "E83C2B2AF3CF011543FBA13FBC524D1122EEA68548F9F27B9F7A82B5D594666C" TREE_HASH_ALGORITHM = "path-nul-bytes-nul-sha256-upper-lf-v1" SOURCE_ARTIFACT_MANIFEST_BYTES = 3770 SOURCE_ARTIFACT_MANIFEST_SHA256 = "132B637634550A43B0EDD5E8E74C439205DE4BFECCE0A9276D8EBEF78445ECFC" FILE_ATTRIBUTE_DIRECTORY = 0x10 FILE_ATTRIBUTE_REPARSE_POINT = 0x400 FILE_READ_ATTRIBUTES = 0x80 GENERIC_READ = 0x80000000 FILE_SHARE_READ = 0x1 FILE_SHARE_WRITE = 0x2 OPEN_EXISTING = 3 FILE_FLAG_OPEN_REPARSE_POINT = 0x00200000 FILE_FLAG_BACKUP_SEMANTICS = 0x02000000 INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value class ValidationError(RuntimeError): def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code @dataclass(frozen=True) class PathIdentity: relative_path: str kind: str volume_serial: int file_id: int attributes: int reparse_tag: int final_path: str @dataclass(frozen=True) class FileEvidence: path: str bytes: int sha256: str class _ByHandleFileInformation(ctypes.Structure): _fields_ = [ ("dwFileAttributes", wintypes.DWORD), ("ftCreationTime", wintypes.FILETIME), ("ftLastAccessTime", wintypes.FILETIME), ("ftLastWriteTime", wintypes.FILETIME), ("dwVolumeSerialNumber", wintypes.DWORD), ("nFileSizeHigh", wintypes.DWORD), ("nFileSizeLow", wintypes.DWORD), ("nNumberOfLinks", wintypes.DWORD), ("nFileIndexHigh", wintypes.DWORD), ("nFileIndexLow", wintypes.DWORD), ] if os.name == "nt": _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) _CreateFileW = _kernel32.CreateFileW _CreateFileW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE, ] _CreateFileW.restype = wintypes.HANDLE _GetFileInformationByHandle = _kernel32.GetFileInformationByHandle _GetFileInformationByHandle.argtypes = [ wintypes.HANDLE, ctypes.POINTER(_ByHandleFileInformation), ] _GetFileInformationByHandle.restype = wintypes.BOOL _GetFinalPathNameByHandleW = _kernel32.GetFinalPathNameByHandleW _GetFinalPathNameByHandleW.argtypes = [ wintypes.HANDLE, wintypes.LPWSTR, wintypes.DWORD, wintypes.DWORD, ] _GetFinalPathNameByHandleW.restype = wintypes.DWORD _CloseHandle = _kernel32.CloseHandle _CloseHandle.argtypes = [wintypes.HANDLE] _CloseHandle.restype = wintypes.BOOL class _HandleSet: def __init__(self) -> None: self.handles: list[int] = [] def add(self, handle: int) -> None: self.handles.append(handle) def close(self) -> None: while self.handles: _CloseHandle(self.handles.pop()) def __enter__(self) -> "_HandleSet": return self def __exit__(self, *_: object) -> None: self.close() def _strict_object(pairs: list[tuple[str, object]]) -> dict[str, object]: result: dict[str, object] = {} for key, value in pairs: if key in result: raise ValidationError("E_JSON", "Duplicate JSON key.") result[key] = value return result def _strict_json(data: bytes) -> dict[str, object]: try: value = json.loads(data.decode("utf-8"), object_pairs_hook=_strict_object) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValidationError("E_JSON", "Invalid strict UTF-8 JSON.") from exc if not isinstance(value, dict): raise ValidationError("E_JSON", "JSON root must be an object.") return value def _validate_project_root_text(value: str) -> Path: if os.name != "nt": raise ValidationError("E_WINDOWS_REQUIRED", "Windows is required.") if not re.fullmatch(r"[A-Za-z]:\\[^:*?\"<>|]+(?:\\[^:*?\"<>|]+)*", value): raise ValidationError("E_PROJECT_ROOT", "Project root must be an unambiguous absolute drive path.") if value.startswith(("\\\\", "\\\\?\\", "\\\\.\\")): raise ValidationError("E_PROJECT_ROOT", "UNC and device paths are forbidden.") components = value[3:].split("\\") if len(value) > 3 else [] if not components or any( part in {"", ".", ".."} or part.endswith((" ", ".")) or ":" in part for part in components ): raise ValidationError("E_PROJECT_ROOT", "Ambiguous project-root components are forbidden.") root = Path(value) if root.name != PROJECT_ID: raise ValidationError("E_PROJECT_ROOT", "Project-root identity mismatch.") return root def _native(relative: PurePosixPath) -> tuple[str, ...]: parts = tuple(relative.parts) if not parts or any(part in {"", ".", ".."} or part.endswith((" ", ".")) or ":" in part for part in parts): raise ValidationError("E_PATH", "Invalid fixed relative path.") return parts def _join(root: Path, relative: PurePosixPath) -> Path: return root.joinpath(*_native(relative)) def _lstat(path: Path) -> os.stat_result: try: return os.lstat(path) except OSError as exc: raise ValidationError("E_PATH_MISSING", "Required lexical path is unavailable.") from exc def _reparse_tag(st: os.stat_result) -> int: return int(getattr(st, "st_reparse_tag", 0) or 0) def _attributes(st: os.stat_result) -> int: return int(getattr(st, "st_file_attributes", 0) or 0) def _open_identity( path: Path, *, relative: str, expect_dir: bool, handles: _HandleSet, ) -> PathIdentity: st = _lstat(path) attrs = _attributes(st) tag = _reparse_tag(st) if attrs & FILE_ATTRIBUTE_REPARSE_POINT or tag or stat.S_ISLNK(st.st_mode): raise ValidationError("E_REPARSE", "Reparse paths are forbidden.") if expect_dir: if not stat.S_ISDIR(st.st_mode): raise ValidationError("E_PATH_KIND", "Expected directory.") access = FILE_READ_ATTRIBUTES share = FILE_SHARE_READ | FILE_SHARE_WRITE else: if not stat.S_ISREG(st.st_mode): raise ValidationError("E_PATH_KIND", "Expected regular file.") access = GENERIC_READ share = FILE_SHARE_READ handle = _CreateFileW( str(path), access, share, None, OPEN_EXISTING, FILE_FLAG_OPEN_REPARSE_POINT | FILE_FLAG_BACKUP_SEMANTICS, None, ) if handle == INVALID_HANDLE_VALUE: raise ValidationError("E_PATH_HANDLE", "Cannot open no-follow path handle.") handle_value = int(handle) handles.add(handle_value) info = _ByHandleFileInformation() if not _GetFileInformationByHandle(handle_value, ctypes.byref(info)): raise ValidationError("E_PATH_HANDLE", "Cannot read path identity.") if info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT: raise ValidationError("E_REPARSE", "Reparse handle is forbidden.") is_directory = bool(info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) if is_directory != expect_dir: raise ValidationError("E_PATH_KIND", "Handle kind mismatch.") size = 512 while True: buffer = ctypes.create_unicode_buffer(size) result = _GetFinalPathNameByHandleW(handle_value, buffer, size, 0) if result == 0: raise ValidationError("E_PATH_HANDLE", "Cannot read final path.") if result < size: final_path = buffer.value break size = result + 1 if final_path.startswith("\\\\?\\UNC\\"): raise ValidationError("E_PATH_ESCAPE", "UNC final paths are forbidden.") if final_path.startswith("\\\\?\\"): final_path = final_path[4:] file_id = (int(info.nFileIndexHigh) << 32) | int(info.nFileIndexLow) return PathIdentity( relative_path=relative, kind="directory" if expect_dir else "file", volume_serial=int(info.dwVolumeSerialNumber), file_id=file_id, attributes=int(info.dwFileAttributes), reparse_tag=tag, final_path=os.path.normpath(final_path), ) def _inside(root_final: str, child_final: str) -> bool: root_norm = os.path.normcase(os.path.normpath(root_final)) child_norm = os.path.normcase(os.path.normpath(child_final)) try: return os.path.commonpath((root_norm, child_norm)) == root_norm except ValueError: return False def _expected_paths(root: Path) -> list[tuple[Path, str, bool]]: paths: dict[str, tuple[Path, str, bool]] = {} def add(path: Path, relative: str, expect_dir: bool) -> None: key = os.path.normcase(os.path.normpath(str(path))) paths[key] = (path, relative, expect_dir) add(root, ".", True) add(_join(root, PROJECT_MARKER_REL), PROJECT_MARKER_REL.as_posix(), False) for relative in ( SOURCE_ROOT_REL, PROJECTION_ROOT_REL, CONTRACT_REL, SOURCE_ARTIFACT_MANIFEST_REL, ): current = root current_parts: list[str] = [] for part in _native(relative): current = current / part current_parts.append(part) relative_text = PurePosixPath(*current_parts).as_posix() is_leaf = len(current_parts) == len(relative.parts) expect_dir = not is_leaf or relative in {SOURCE_ROOT_REL, PROJECTION_ROOT_REL} add(current, relative_text, expect_dir) for parent in (SOURCE_ROOT_REL, PROJECTION_ROOT_REL): for name in EXPECTED_FILES: add(_join(root, parent) / name, (parent / name).as_posix(), False) return sorted(paths.values(), key=lambda item: (len(Path(item[0]).parts), item[1])) def _capture(root: Path) -> tuple[dict[str, PathIdentity], dict[str, FileEvidence], dict[str, object]]: with _HandleSet() as handles: identities: dict[str, PathIdentity] = {} for path, relative, expect_dir in _expected_paths(root): identity = _open_identity( path, relative=relative, expect_dir=expect_dir, handles=handles, ) identities[relative] = identity root_identity = identities["."] if root_identity.final_path != os.path.normpath(str(root)): raise ValidationError("E_PATH_CASE_DRIFT", "Project-root physical casing mismatch.") if os.path.normcase(root_identity.final_path) != os.path.normcase(os.path.normpath(str(root))): raise ValidationError("E_PATH_ESCAPE", "Project-root physical identity mismatch.") for identity in identities.values(): if identity.volume_serial != root_identity.volume_serial or not _inside( root_identity.final_path, identity.final_path ): raise ValidationError("E_PATH_ESCAPE", "Path escapes the trusted physical project root.") expected_final = root_identity.final_path if identity.relative_path != ".": expected_final = os.path.normpath( os.path.join(root_identity.final_path, *PurePosixPath(identity.relative_path).parts) ) if identity.final_path != expected_final: raise ValidationError("E_PATH_CASE_DRIFT", "Actual NTFS path casing does not match the frozen path.") marker_path = _join(root, PROJECT_MARKER_REL) marker = marker_path.read_text(encoding="utf-8") if not re.search(r"(?m)^project:\s*$", marker) or not re.search( rf"(?m)^ id: {re.escape(PROJECT_ID)}\s*$", marker ): raise ValidationError("E_PROJECT_ROOT", "Project marker mismatch.") contract_path = _join(root, CONTRACT_REL) contract = _strict_json(contract_path.read_bytes()) _validate_contract_shape(contract) source_artifact_manifest = _join(root, SOURCE_ARTIFACT_MANIFEST_REL).read_bytes() _validate_source_artifact_manifest(source_artifact_manifest, contract) source_root = _join(root, SOURCE_ROOT_REL) projection_root = _join(root, PROJECTION_ROOT_REL) _validate_projection_set(projection_root) source_evidence = _read_files(source_root) projection_evidence = _read_files(projection_root) _validate_evidence(contract, source_evidence, projection_evidence) _validate_manifest((projection_root / "manifest.json").read_bytes()) return identities, projection_evidence, contract def _validate_projection_set(root: Path) -> None: actual: list[str] = [] try: with os.scandir(root) as entries: for entry in entries: if entry.name.startswith("_"): raise ValidationError("E_RESERVED_NAME", "Reserved underscore path is forbidden.") if entry.is_symlink() or not entry.is_file(follow_symlinks=False): raise ValidationError("E_PROJECTION_SET", "Projection must contain only regular files.") actual.append(entry.name) except OSError as exc: raise ValidationError("E_PROJECTION_SET", "Cannot enumerate projection.") from exc if tuple(sorted(actual)) != EXPECTED_FILES: raise ValidationError("E_PROJECTION_SET", "Projection exact file set mismatch.") def _read_files(root: Path) -> dict[str, FileEvidence]: result: dict[str, FileEvidence] = {} for name in EXPECTED_FILES: data = (root / name).read_bytes() result[name] = FileEvidence(name, len(data), hashlib.sha256(data).hexdigest().upper()) return result def _validate_contract_shape(contract: dict[str, object]) -> None: expected_keys = { "schema", "task_id", "project_id", "source_root", "projection_root", "expected_extension_id", "public_key_der_sha256", "tree_hash_algorithm", "tree_sha256", "files", } if set(contract) != expected_keys: raise ValidationError("E_CONTRACT", "Contract key set mismatch.") fixed = { "schema": SCHEMA, "task_id": TASK_ID, "project_id": PROJECT_ID, "source_root": SOURCE_ROOT_REL.as_posix(), "projection_root": PROJECTION_ROOT_REL.as_posix(), "expected_extension_id": EXPECTED_EXTENSION_ID, "public_key_der_sha256": EXPECTED_PUBLIC_DER_SHA256, "tree_hash_algorithm": TREE_HASH_ALGORITHM, } if any(contract.get(key) != value for key, value in fixed.items()): raise ValidationError("E_CONTRACT", "Contract fixed identity mismatch.") files = contract.get("files") if not isinstance(files, list) or len(files) != len(EXPECTED_FILES): raise ValidationError("E_CONTRACT", "Contract file set mismatch.") def _tree_hash(evidence: Iterable[FileEvidence]) -> str: payload = bytearray() for item in sorted(evidence, key=lambda value: value.path): payload.extend(item.path.encode("utf-8")) payload.append(0) payload.extend(str(item.bytes).encode("ascii")) payload.append(0) payload.extend(item.sha256.upper().encode("ascii")) payload.append(0x0A) return hashlib.sha256(payload).hexdigest().upper() def _contract_file_evidence(contract: dict[str, object]) -> dict[str, FileEvidence]: files = contract["files"] assert isinstance(files, list) expected: dict[str, FileEvidence] = {} for raw in files: if not isinstance(raw, dict) or set(raw) != {"path", "bytes", "sha256"}: raise ValidationError("E_CONTRACT", "Contract file entry mismatch.") path = raw.get("path") size = raw.get("bytes") digest = raw.get("sha256") if ( not isinstance(path, str) or path not in EXPECTED_FILES or isinstance(size, bool) or not isinstance(size, int) or size <= 0 or not isinstance(digest, str) or not re.fullmatch(r"[0-9A-F]{64}", digest) or path in expected ): raise ValidationError("E_CONTRACT", "Invalid contract file entry.") expected[path] = FileEvidence(path, size, digest) if tuple(sorted(expected)) != EXPECTED_FILES: raise ValidationError("E_CONTRACT", "Contract file paths mismatch.") return expected def _validate_source_artifact_manifest(data: bytes, contract: dict[str, object]) -> None: if ( len(data) != SOURCE_ARTIFACT_MANIFEST_BYTES or hashlib.sha256(data).hexdigest().upper() != SOURCE_ARTIFACT_MANIFEST_SHA256 ): raise ValidationError("E_SOURCE_MANIFEST", "Frozen source artifact manifest bytes mismatch.") source_manifest = _strict_json(data) files = source_manifest.get("files") if not isinstance(files, list): raise ValidationError("E_SOURCE_MANIFEST", "Source artifact manifest files are invalid.") expected = _contract_file_evidence(contract) observed: dict[str, FileEvidence] = {} for raw in files: if not isinstance(raw, dict): raise ValidationError("E_SOURCE_MANIFEST", "Source artifact manifest entry is invalid.") path = raw.get("path") if path not in EXPECTED_FILES: continue if set(raw) != {"path", "bytes", "sha256"} or path in observed: raise ValidationError("E_SOURCE_MANIFEST", "Source artifact manifest exact-five entry is invalid.") size = raw.get("bytes") digest = raw.get("sha256") if isinstance(size, bool) or not isinstance(size, int) or not isinstance(digest, str): raise ValidationError("E_SOURCE_MANIFEST", "Source artifact manifest exact-five value is invalid.") observed[path] = FileEvidence(path, size, digest) if observed != expected: raise ValidationError("E_SOURCE_MANIFEST", "Source artifact manifest exact-five binding mismatch.") def _validate_evidence( contract: dict[str, object], source: dict[str, FileEvidence], projection: dict[str, FileEvidence], ) -> None: expected = _contract_file_evidence(contract) if source != expected or projection != expected or source != projection: raise ValidationError("E_BYTE_DRIFT", "Source/projection bytes are not exactly locked.") tree = _tree_hash(projection.values()) if contract.get("tree_sha256") != tree: raise ValidationError("E_TREE_HASH", "Projection tree hash mismatch.") def _validate_manifest(data: bytes) -> None: manifest = _strict_json(data) if ( manifest.get("manifest_version") != 3 or manifest.get("name") != "project-info Bilibili 完整视频入口" or manifest.get("version") != "1.2.25" or manifest.get("version_name") != "1.2.25+20260829.generic.v027" or manifest.get("background") != {"service_worker": "background.js", "type": "module"} or manifest.get("permissions") != ["alarms", "cookies", "nativeMessaging", "scripting", "storage", "tabs"] or manifest.get("host_permissions") != [ "https://www.bilibili.com/*", "https://*.bilibili.com/*", ] or "side_panel" in manifest or "action" in manifest ): raise ValidationError("E_MANIFEST", "Manifest identity mismatch.") key = manifest.get("key") if not isinstance(key, str): raise ValidationError("E_MANIFEST", "Manifest public key is missing.") try: der = base64.b64decode(key, validate=True) except ValueError as exc: raise ValidationError("E_MANIFEST", "Manifest public key is invalid.") from exc digest = hashlib.sha256(der).hexdigest().upper() if digest != EXPECTED_PUBLIC_DER_SHA256: raise ValidationError("E_MANIFEST", "Manifest public key hash mismatch.") extension_id = "".join(chr(ord("a") + int(nibble, 16)) for nibble in digest[:32].lower()) if extension_id != EXPECTED_EXTENSION_ID: raise ValidationError("E_EXTENSION_ID", "Derived extension ID mismatch.") def _identity_key(value: PathIdentity) -> tuple[object, ...]: return ( value.relative_path, value.kind, value.volume_serial, value.file_id, value.attributes, value.reparse_tag, value.final_path, ) def validate_project( project_root_text: str, *, between_passes: Callable[[], None] | None = None, ) -> dict[str, object]: root = _validate_project_root_text(project_root_text) first_identities, first_files, first_contract = _capture(root) if between_passes is not None: between_passes() second_identities, second_files, second_contract = _capture(root) if set(first_identities) != set(second_identities) or any( _identity_key(first_identities[key]) != _identity_key(second_identities[key]) for key in first_identities ): raise ValidationError("E_PATH_IDENTITY_DRIFT", "Path identity changed between validation passes.") if first_files != second_files or first_contract != second_contract: raise ValidationError("E_PATH_IDENTITY_DRIFT", "Content identity changed between validation passes.") root_identity = second_identities["."] return { "schema": 1, "status": "VALIDATION_PASS_ONLY", "project_id": PROJECT_ID, "task_id": TASK_ID, "project_root": str(root), "project_root_final": root_identity.final_path, "project_volume_serial": root_identity.volume_serial, "project_file_id": root_identity.file_id, "source_root": SOURCE_ROOT_REL.as_posix(), "projection_root": PROJECTION_ROOT_REL.as_posix(), "file_count": len(EXPECTED_FILES), "tree_sha256": second_contract["tree_sha256"], "extension_id": EXPECTED_EXTENSION_ID, "source_artifact_manifest": { "bytes": SOURCE_ARTIFACT_MANIFEST_BYTES, "sha256": SOURCE_ARTIFACT_MANIFEST_SHA256, }, "files": [asdict(second_files[name]) for name in EXPECTED_FILES], "path_identities": [asdict(second_identities[key]) for key in sorted(second_identities)], } def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Validate the exact local-unpacked Bilibili extension projection without launching Chrome.", ) parser.add_argument("--project-root", required=True, help="Exact absolute trusted project root.") return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) try: result = validate_project(args.project_root) except ValidationError as exc: print(json.dumps({"schema": 1, "status": "SAFETY_STOP", "error_code": exc.code}, separators=(",", ":"))) return 3 print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())