from __future__ import annotations import base64 import hashlib import io import json import pathlib import struct import sys import subprocess import unittest 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 ( # noqa: E402 CANONICAL_URL, EXPECTED_EXTENSION_ID, EXPECTED_ORIGIN, EXTENSION_BUILD, MANIFEST_PUBLIC_KEY, PUBLIC_KEY_DER_SHA256, TARGET_BVID, ) from bili_authenticated_extension.protocol import ( # noqa: E402 ProtocolError, read_frame, strict_json_loads, validate_cancel, validate_hello, validate_origin_argv, validate_prepare, validate_start, ) ROOT = PROJECT_DEV / "bili_authenticated_extension" def valid_start(now_ms: int = 2_000_000_000_000) -> dict: return { "schema": 2, "type": "start", "extension_build": EXTENSION_BUILD, "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "cookie_store_id": "0", "prepare_id": "b" * 32, "page_proof": { "target": TARGET_BVID, "canonical_url": CANONICAL_URL, "task_nonce": "a" * 32, "observed_at_unix_ms": now_ms, "observed_duration_ms": 3_133_950, "video_width": 1920, "video_height": 1080, "ready_state": 4, "eme_present": False, }, "cookies": [ { "name": "synthetic_name", "value": "synthetic_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 ProtocolTests(unittest.TestCase): def test_public_key_recomputes_hash_and_extension_id(self) -> None: der = base64.b64decode(MANIFEST_PUBLIC_KEY, validate=True) digest = hashlib.sha256(der).hexdigest().upper() self.assertEqual(PUBLIC_KEY_DER_SHA256, digest) alphabet = "abcdefghijklmnop" extension_id = "".join(alphabet[nibble] for byte in bytes.fromhex(digest[:32]) for nibble in (byte >> 4, byte & 15)) self.assertEqual(EXPECTED_EXTENSION_ID, extension_id) def test_manifest_permissions_and_native_origin_are_exact(self) -> None: manifest = json.loads((ROOT / "manifest.json").read_text(encoding="utf-8")) self.assertEqual(MANIFEST_PUBLIC_KEY, manifest["key"]) self.assertEqual( {"activeTab", "cookies", "nativeMessaging", "scripting", "sidePanel"}, set(manifest["permissions"]), ) self.assertEqual(["https://www.bilibili.com/*"], manifest["host_permissions"]) for forbidden in ("downloads", "storage", "debugger", "webRequest", "content_scripts", "externally_connectable"): self.assertNotIn(forbidden, json.dumps(manifest)) native = json.loads((ROOT / "native-host-manifest.template.json").read_text(encoding="utf-8")) self.assertEqual([EXPECTED_ORIGIN], native["allowed_origins"]) def test_source_artifact_manifest_binds_every_product_file(self) -> None: artifact_path = ROOT / "source-artifact-manifest.json" artifact = json.loads(artifact_path.read_text(encoding="utf-8")) self.assertEqual(EXPECTED_EXTENSION_ID, artifact["extension_id"]) expected_names = { item.relative_to(ROOT).as_posix() for item in ROOT.rglob("*") if item.is_file() and item != artifact_path } self.assertEqual(expected_names, {entry["path"] for entry in artifact["files"]}) for entry in artifact["files"]: path = ROOT / entry["path"] payload = path.read_bytes() self.assertEqual(entry["bytes"], len(payload), entry["path"]) self.assertEqual(entry["sha256"], hashlib.sha256(payload).hexdigest().upper(), entry["path"]) dependency_path = ROOT / "dependencies" / "dependency-artifact-manifest.json" dependency_payload = dependency_path.read_bytes() self.assertEqual(len(dependency_payload), artifact["dependency_artifact_manifest_bytes"]) self.assertEqual( hashlib.sha256(dependency_payload).hexdigest().upper(), artifact["dependency_artifact_manifest_sha256"], ) dependency = json.loads(dependency_payload.decode("utf-8")) self.assertEqual("yt-dlp", dependency["distribution"]) self.assertEqual("2026.7.4", dependency["version"]) self.assertEqual([], dependency["unconditional_runtime_dependencies"]) self.assertEqual(1, len(dependency["wheels"])) def test_origin_and_handshake_exact(self) -> None: validate_origin_argv([EXPECTED_ORIGIN], EXPECTED_ORIGIN) validate_origin_argv([EXPECTED_ORIGIN, "--parent-window=42"], EXPECTED_ORIGIN) for args in ([], ["chrome-extension://wrong/"], [EXPECTED_ORIGIN, "--bad"], [EXPECTED_ORIGIN, "x", "y"]): with self.subTest(args=args), self.assertRaises(ProtocolError): validate_origin_argv(list(args), EXPECTED_ORIGIN) validate_hello({"schema": 2, "type": "hello", "extension_build": EXTENSION_BUILD, "target": TARGET_BVID}) def test_start_and_cookie_contract(self) -> None: baseline = valid_start() prepare = { "schema": 2, "type": "prepare", "extension_build": EXTENSION_BUILD, "target": TARGET_BVID, "prepare_id": "b" * 32, "page_proof": dict(baseline["page_proof"]), } self.assertIs(prepare, validate_prepare(prepare, now_ms=2_000_000_000_000)) with self.assertRaises(ProtocolError): validate_prepare(dict(prepare, cookies=[]), now_ms=2_000_000_000_000) with self.assertRaises(ProtocolError): validate_prepare(dict(prepare, prepare_id="wrong"), now_ms=2_000_000_000_000) self.assertIs(baseline, validate_start(baseline, now_ms=2_000_000_000_000)) mutations = [] for key, value in ( ("target", "BV1WRONG"), ("canonical_url", "https://example.invalid/"), ("cookie_store_id", "x"), ("prepare_id", "wrong"), ): changed = valid_start() changed[key] = value mutations.append(changed) for key, value in ( ("observed_duration_ms", 600_133), ("ready_state", True), ("eme_present", True), ("video_width", 0), ("observed_at_unix_ms", 1_999_999_900_000), ): changed = valid_start() changed["page_proof"][key] = value mutations.append(changed) for key, value in ( ("domain", ".example.com"), ("partition_key", {"topLevelSite": "https://example.invalid"}), ("store_id", "1"), ("value", ""), ("path", "/wrong"), ("expiration_unix", 3), ): changed = valid_start() changed["cookies"][0][key] = value mutations.append(changed) changed = valid_start() changed["unknown"] = 1 mutations.append(changed) for changed in mutations: with self.subTest(changed=changed), self.assertRaises(ProtocolError): validate_start(changed, now_ms=2_000_000_000_000) def test_duplicate_nonfinite_and_oversized_frames_fail(self) -> None: with self.assertRaises(ProtocolError): strict_json_loads(b'{"schema":2,"schema":2}') with self.assertRaises(ProtocolError): strict_json_loads(b'{"x":NaN}') with self.assertRaises(ProtocolError): read_frame(io.BytesIO(struct.pack(" None: value = {"schema": 2, "type": "cancel", "target": TARGET_BVID, "task_nonce": "0" * 32} validate_cancel(value) changed = dict(value, cookie="synthetic") with self.assertRaises(ProtocolError): validate_cancel(changed) def test_extension_sources_have_no_forbidden_control_surface(self) -> None: sources = "\n".join((ROOT / name).read_text(encoding="utf-8") for name in ("background.js", "sidepanel.js")) for forbidden in ( "chrome.downloads", "chrome.storage", "externally_connectable", "127.0.0.1", "localhost", "cookies.getAll({domain", ): self.assertNotIn(forbidden, sources) self.assertIn('chrome.cookies.getAll({url: "https://www.bilibili.com/", storeId})', sources) self.assertIn("crypto.getRandomValues", sources) def test_mock_chrome_requires_two_clicks_and_rereads_on_retry(self) -> None: node_test = pathlib.Path(__file__).with_name("background_contract.mjs") result = subprocess.run( ["node", str(node_test), str(ROOT / "background.js")], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("BACKGROUND_CONTRACT_PASS", result.stdout.strip()) if __name__ == "__main__": unittest.main()