"""Strict Native Messaging wire validation.
|
|
This module is stdlib-only and contains no downloader imports.
|
"""
|
|
from __future__ import annotations
|
|
import json
|
import re
|
import struct
|
import time
|
import unicodedata
|
from typing import Any, BinaryIO, Iterable
|
|
from .constants import (
|
CANONICAL_URL,
|
DURATION_TOLERANCE_MS,
|
EXPECTED_DURATION_MS,
|
EXTENSION_BUILD,
|
HOST_BUILD,
|
MAX_COOKIE_COUNT,
|
MAX_INPUT_FRAME,
|
MAX_OUTPUT_FRAME,
|
MAX_SAFE_INTEGER,
|
SCHEMA_VERSION,
|
TARGET_BVID,
|
TARGET_PATH,
|
)
|
|
_NONCE_RE = re.compile(r"[0-9a-f]{32}\Z")
|
_PREPARE_ID_RE = re.compile(r"[0-9a-f]{32}\Z")
|
_STORE_RE = re.compile(r"[0-9]{1,8}\Z")
|
_PARENT_RE = re.compile(r"--parent-window=[0-9]+\Z")
|
_COOKIE_SAME_SITE = {"no_restriction", "lax", "strict", "unspecified"}
|
|
|
class ProtocolError(Exception):
|
"""A fixed-code input rejection that is safe to expose."""
|
|
def __init__(self, code: str = "E_PROTOCOL") -> None:
|
super().__init__(code)
|
self.code = code
|
|
|
def _reject_constant(_: str) -> None:
|
raise ProtocolError()
|
|
|
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
result: dict[str, Any] = {}
|
for key, value in pairs:
|
if key in result:
|
raise ProtocolError()
|
result[key] = value
|
return result
|
|
|
def strict_json_loads(payload: bytes) -> dict[str, Any]:
|
try:
|
text = payload.decode("utf-8", errors="strict")
|
value = json.loads(
|
text,
|
object_pairs_hook=_unique_object,
|
parse_constant=_reject_constant,
|
)
|
except ProtocolError:
|
raise
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
raise ProtocolError() from exc
|
if not isinstance(value, dict):
|
raise ProtocolError()
|
return value
|
|
|
def encode_json(value: dict[str, Any], limit: int = MAX_OUTPUT_FRAME) -> bytes:
|
payload = json.dumps(
|
value,
|
ensure_ascii=True,
|
allow_nan=False,
|
separators=(",", ":"),
|
sort_keys=True,
|
).encode("utf-8")
|
if len(payload) > limit:
|
raise ProtocolError()
|
return payload
|
|
|
def read_frame(stream: BinaryIO, limit: int = MAX_INPUT_FRAME) -> bytes | None:
|
header = stream.read(4)
|
if header == b"":
|
return None
|
if len(header) != 4:
|
raise ProtocolError()
|
(length,) = struct.unpack("<I", header)
|
if length == 0 or length > limit:
|
raise ProtocolError()
|
payload = stream.read(length)
|
if len(payload) != length:
|
raise ProtocolError()
|
return payload
|
|
|
def write_frame(stream: BinaryIO, value: dict[str, Any]) -> None:
|
payload = encode_json(value)
|
stream.write(struct.pack("<I", len(payload)))
|
stream.write(payload)
|
stream.flush()
|
|
|
def _exact_keys(value: dict[str, Any], expected: Iterable[str]) -> None:
|
if set(value) != set(expected):
|
raise ProtocolError()
|
|
|
def _integer(value: Any, minimum: int, maximum: int) -> int:
|
if isinstance(value, bool) or not isinstance(value, int):
|
raise ProtocolError()
|
if not minimum <= value <= maximum:
|
raise ProtocolError()
|
return value
|
|
|
def _boolean(value: Any) -> bool:
|
if not isinstance(value, bool):
|
raise ProtocolError()
|
return value
|
|
|
def _safe_string(value: Any, minimum: int, maximum_utf8: int) -> str:
|
if not isinstance(value, str):
|
raise ProtocolError()
|
encoded = value.encode("utf-8")
|
if len(encoded) < minimum or len(encoded) > maximum_utf8:
|
raise ProtocolError()
|
if any(ch in "\x00\t\r\n" or unicodedata.category(ch).startswith("C") for ch in value):
|
raise ProtocolError()
|
return value
|
|
|
def validate_origin_argv(arguments: list[str], expected_origin: str) -> None:
|
if len(arguments) not in (1, 2) or arguments[0] != expected_origin:
|
raise ProtocolError("E_ORIGIN")
|
if len(arguments) == 2 and not _PARENT_RE.fullmatch(arguments[1]):
|
raise ProtocolError("E_ORIGIN")
|
|
|
def _validate_common(value: dict[str, Any], message_type: str, keys: Iterable[str]) -> None:
|
_exact_keys(value, keys)
|
if _integer(value.get("schema"), SCHEMA_VERSION, SCHEMA_VERSION) != SCHEMA_VERSION:
|
raise ProtocolError()
|
if value.get("type") != message_type or value.get("target") != TARGET_BVID:
|
raise ProtocolError()
|
|
|
def validate_hello(value: dict[str, Any]) -> dict[str, Any]:
|
_validate_common(value, "hello", {"schema", "type", "extension_build", "target"})
|
if value["extension_build"] != EXTENSION_BUILD:
|
raise ProtocolError("E_BUILD")
|
return value
|
|
|
def validate_status(value: dict[str, Any]) -> dict[str, Any]:
|
_validate_common(value, "status", {"schema", "type", "target"})
|
return value
|
|
|
def validate_cancel(value: dict[str, Any]) -> dict[str, Any]:
|
_validate_common(value, "cancel", {"schema", "type", "target", "task_nonce"})
|
if not isinstance(value["task_nonce"], str) or not _NONCE_RE.fullmatch(value["task_nonce"]):
|
raise ProtocolError()
|
return value
|
|
|
def validate_prepare(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
|
_validate_common(
|
value,
|
"prepare",
|
{"schema", "type", "extension_build", "target", "prepare_id", "page_proof"},
|
)
|
if value["extension_build"] != EXTENSION_BUILD:
|
raise ProtocolError("E_BUILD")
|
if not isinstance(value["prepare_id"], str) or not _PREPARE_ID_RE.fullmatch(value["prepare_id"]):
|
raise ProtocolError("E_PREPARE")
|
validate_page_proof(value["page_proof"], now_ms=now_ms)
|
return value
|
|
|
def validate_page_proof(value: Any, now_ms: int | None = None) -> dict[str, Any]:
|
if not isinstance(value, dict):
|
raise ProtocolError()
|
_exact_keys(
|
value,
|
{
|
"target",
|
"canonical_url",
|
"task_nonce",
|
"observed_at_unix_ms",
|
"observed_duration_ms",
|
"video_width",
|
"video_height",
|
"ready_state",
|
"eme_present",
|
},
|
)
|
if value["target"] != TARGET_BVID or value["canonical_url"] != CANONICAL_URL:
|
raise ProtocolError()
|
if not isinstance(value["task_nonce"], str) or not _NONCE_RE.fullmatch(value["task_nonce"]):
|
raise ProtocolError()
|
observed_at = _integer(value["observed_at_unix_ms"], 1, MAX_SAFE_INTEGER)
|
observed_duration = _integer(value["observed_duration_ms"], 1, MAX_SAFE_INTEGER)
|
_integer(value["video_width"], 1, 7680)
|
_integer(value["video_height"], 1, 4320)
|
_integer(value["ready_state"], 1, 4)
|
if _boolean(value["eme_present"]):
|
raise ProtocolError("E_DRM")
|
if abs(observed_duration - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
|
raise ProtocolError("E_DURATION")
|
current = int(time.time() * 1000) if now_ms is None else now_ms
|
if observed_at < current - 60_000 or observed_at > current + 5_000:
|
raise ProtocolError("E_PAGE_PROOF")
|
return value
|
|
|
def _cookie_path_matches(path: str) -> bool:
|
if path == "/":
|
return True
|
if not TARGET_PATH.startswith(path):
|
return False
|
return path.endswith("/") or len(path) == len(TARGET_PATH) or TARGET_PATH[len(path)] == "/"
|
|
|
def validate_cookie(
|
value: Any,
|
store_id: str,
|
observed_at_unix_ms: int,
|
) -> dict[str, Any]:
|
if not isinstance(value, dict):
|
raise ProtocolError("E_SECRET_INPUT")
|
try:
|
_exact_keys(
|
value,
|
{
|
"name",
|
"value",
|
"domain",
|
"host_only",
|
"path",
|
"secure",
|
"http_only",
|
"same_site",
|
"session",
|
"expiration_unix",
|
"store_id",
|
"partition_key",
|
},
|
)
|
_safe_string(value["name"], 1, 256)
|
_safe_string(value["value"], 1, 4096)
|
host_only = _boolean(value["host_only"])
|
expected_domain = "www.bilibili.com" if host_only else ".bilibili.com"
|
if value["domain"] != expected_domain:
|
raise ProtocolError()
|
path = _safe_string(value["path"], 1, 1024)
|
if not path.startswith("/") or not _cookie_path_matches(path):
|
raise ProtocolError()
|
_boolean(value["secure"])
|
_boolean(value["http_only"])
|
session = _boolean(value["session"])
|
if value["same_site"] not in _COOKIE_SAME_SITE:
|
raise ProtocolError()
|
if value["store_id"] != store_id or value["partition_key"] is not None:
|
raise ProtocolError()
|
if session:
|
if value["expiration_unix"] is not None:
|
raise ProtocolError()
|
else:
|
expires = _integer(value["expiration_unix"], 1, MAX_SAFE_INTEGER)
|
if expires <= observed_at_unix_ms // 1000:
|
raise ProtocolError()
|
except ProtocolError as exc:
|
raise ProtocolError("E_SECRET_INPUT") from exc
|
return value
|
|
|
def validate_start(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
|
_validate_common(
|
value,
|
"start",
|
{
|
"schema",
|
"type",
|
"extension_build",
|
"target",
|
"canonical_url",
|
"cookie_store_id",
|
"prepare_id",
|
"page_proof",
|
"cookies",
|
},
|
)
|
if value["extension_build"] != EXTENSION_BUILD:
|
raise ProtocolError("E_BUILD")
|
if value["canonical_url"] != CANONICAL_URL:
|
raise ProtocolError()
|
if not isinstance(value["prepare_id"], str) or not _PREPARE_ID_RE.fullmatch(value["prepare_id"]):
|
raise ProtocolError("E_PREPARE")
|
store_id = value["cookie_store_id"]
|
if not isinstance(store_id, str) or not _STORE_RE.fullmatch(store_id):
|
raise ProtocolError("E_SECRET_INPUT")
|
proof = validate_page_proof(value["page_proof"], now_ms=now_ms)
|
cookies = value["cookies"]
|
if not isinstance(cookies, list) or not 1 <= len(cookies) <= MAX_COOKIE_COUNT:
|
raise ProtocolError("E_SECRET_INPUT")
|
for cookie in cookies:
|
validate_cookie(cookie, store_id, proof["observed_at_unix_ms"])
|
return value
|
|
|
def validate_message(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
|
message_type = value.get("type")
|
if message_type == "hello":
|
return validate_hello(value)
|
if message_type == "status":
|
return validate_status(value)
|
if message_type == "cancel":
|
return validate_cancel(value)
|
if message_type == "prepare":
|
return validate_prepare(value, now_ms=now_ms)
|
if message_type == "start":
|
return validate_start(value, now_ms=now_ms)
|
raise ProtocolError()
|
|
|
def safe_response(
|
message_type: str,
|
phase: str,
|
*,
|
progress: int = 0,
|
error_code: str | None = None,
|
formal_filename: str | None = None,
|
mapping_filename: str | None = None,
|
prepare_id: str | None = None,
|
) -> dict[str, Any]:
|
return {
|
"schema": SCHEMA_VERSION,
|
"type": message_type,
|
"host_build": HOST_BUILD,
|
"target": TARGET_BVID,
|
"phase": phase,
|
"progress": max(0, min(100, int(progress))),
|
"error_code": error_code,
|
"formal_filename": formal_filename,
|
"mapping_filename": mapping_filename,
|
"prepare_id": prepare_id,
|
}
|
