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("<I", 131_073)), 131_072)
|
with self.assertRaises(ProtocolError):
|
read_frame(io.BytesIO(struct.pack("<I", 4) + b"xx"), 131_072)
|
|
def test_cancel_has_no_secret_fields(self) -> 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()
|
