chore(project-info): archive Bilibili dynamic 344d583568e4

MB-X Bilibili Pipeline

6 days ago af8d9f9dc4d7df459c112d72c14daa3879914411

 dev/project-dev/bili_dynamic_refresh.py

@@ -14,12 +14,12 @@
import math
import os
import re
import secrets
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Iterable, Mapping, Sequence
from urllib.parse import urlsplit

import bili_dynamic_collector as core

@@ -171,48 +171,6 @@
    return core.canonical_json_bytes(signable, newline=False)


def _attest_controller_evidence(
    pending: Mapping[str, Any],
    evidence: dict[str, Any],
    *,
    action_dispatched: bool,
    monotonic_run_started_ms: int,
    monotonic_action_started_ms: int | None,
    monotonic_action_finished_ms: int | None,
    monotonic_observation_started_ms: int | None,
    monotonic_observation_finished_ms: int | None,
    monotonic_evidence_write_started_ms: int,
) -> dict[str, Any]:
    """Bind evidence to the source-controlled controller's actual call envelope.

    The per-run capability is durable state that is never returned by
    ``refresh-begin``.  It is not browser/session material; it only prevents a
    caller-authored JSON document from selecting a successful runtime state.
    """
    capability = pending.get("controller_capability")
    if not isinstance(capability, str) or re.fullmatch(r"[0-9a-f]{64}", capability) is None:
        raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Controller capability is invalid.", safety=True)
    runtime = pending["runtime_contract"]
    evidence["controller_attestation"] = {
        "controller_id": runtime["controller_id"],
        "controller_sha256": runtime["controller_sha256"],
        "binding_algorithm": runtime["binding_algorithm"],
        "action_dispatched": action_dispatched,
        "monotonic_run_started_ms": monotonic_run_started_ms,
        "monotonic_action_started_ms": monotonic_action_started_ms,
        "monotonic_action_finished_ms": monotonic_action_finished_ms,
        "monotonic_observation_started_ms": monotonic_observation_started_ms,
        "monotonic_observation_finished_ms": monotonic_observation_finished_ms,
        "monotonic_evidence_write_started_ms": monotonic_evidence_write_started_ms,
        "binding_sha256": None,
    }
    digest = hmac.new(
        bytes.fromhex(capability), _controller_attestation_payload(evidence), hashlib.sha256
    ).hexdigest()
    evidence["controller_attestation"]["binding_sha256"] = digest
    return evidence


def _strict_json_bytes(payload: bytes, description: str) -> Any:
    if payload.startswith(b"\xef\xbb\xbf"):
        raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} must not contain a BOM.")
@@ -316,11 +274,11 @@
        "intake_root", "evidence_identity", "planned_terminal", "transaction_identity",
        "last_transition_at",
        "runtime_contract",
        "controller_capability",
        "controller_key_commitment", "controller_binding_sha256",
    }
    schema = value.get("schema_version")
    if schema == LEGACY_PENDING_SCHEMA:
        legacy_required = required - {"runtime_contract", "controller_capability"}
        legacy_required = required - {"runtime_contract", "controller_key_commitment", "controller_binding_sha256"}
        if set(value) != legacy_required:
            raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Legacy refresh pending schema is invalid.", safety=True)
    elif schema != PENDING_SCHEMA or set(value) != required:
@@ -449,13 +407,28 @@
        raise core.CollectorError("E_RUN_HOUR_OCCUPIED", "Run slot is not safely reusable.", safety=True)


def refresh_begin(config: core.CollectorConfig, config_path: Path, now: datetime) -> dict[str, Any]:
def refresh_begin(
    config: core.CollectorConfig,
    config_path: Path,
    now: datetime,
    *,
    _controller_key_commitment: str | None = None,
) -> dict[str, Any]:
    refresh = config.refresh
    assert refresh is not None
    _validate_refresh_roots(config, create_state=True)
    recovered = _recover_or_replay(config, config_path, now)
    if recovered is not None:
        return recovered
    if (
        not isinstance(_controller_key_commitment, str)
        or core.LOWER_SHA256_PATTERN.fullmatch(_controller_key_commitment) is None
    ):
        raise core.CollectorError(
            "E_CONTROLLER_ENTRY_REQUIRED",
            "New runtime-v2 runs must be created by the source-controlled refresh-run entry.",
            safety=True,
        )
    hour_epoch = math.floor(now.timestamp() / 3600)
    _validate_slot_available(config, hour_epoch)
    run_id = hashlib.sha256(
@@ -487,7 +460,8 @@
        "transaction_identity": None,
        "last_transition_at": core.canonical_datetime(now),
        "runtime_contract": runtime_identity,
        "controller_capability": secrets.token_hex(32),
        "controller_key_commitment": _controller_key_commitment,
        "controller_binding_sha256": None,
    }
    _write_pending(config, pending, create=True)
    _ensure_started_slot(config, pending)
@@ -719,6 +693,8 @@
    config: core.CollectorConfig,
    pending: Mapping[str, Any],
    path: Path,
    *,
    _controller_key: bytes | None = None,
) -> tuple[dict[str, Any], bytes, dict[str, Any] | None, bool]:
    global _OBSERVATION_CONFIG
    if str(path) != pending["evidence_path"]:
@@ -752,12 +728,24 @@
    binding = attestation["binding_sha256"]
    if not isinstance(binding, str) or core.LOWER_SHA256_PATTERN.fullmatch(binding) is None:
        raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller binding is invalid.", safety=True)
    capability = pending["controller_capability"]
    expected_binding = hmac.new(
        bytes.fromhex(capability), _controller_attestation_payload(value), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(binding, expected_binding):
        raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller envelope binding mismatch.", safety=True)
    if _controller_key is not None:
        commitment = hashlib.sha256(_controller_key).hexdigest()
        if not hmac.compare_digest(commitment, pending["controller_key_commitment"]):
            raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller authority commitment mismatch.", safety=True)
        expected_binding = hmac.new(
            _controller_key, _controller_attestation_payload(value), hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(binding, expected_binding):
            raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller envelope binding mismatch.", safety=True)
    elif (
        pending.get("phase") != "EVIDENCE_BOUND"
        or pending.get("controller_binding_sha256") != binding
    ):
        raise core.CollectorError(
            "E_CONTROLLER_REQUIRED",
            "Caller-authored runtime-v2 evidence cannot enter the commit path.",
            safety=True,
        )
    runtime_contract = _exact_keys(
        value["runtime_contract"], {"contract_id", "contract_bytes", "contract_sha256"}, "runtime_contract"
    )
@@ -944,45 +932,39 @@
    return value, payload, observation_result, identity_match


def bind_controller_evidence(
    config: core.CollectorConfig,
    pending: dict[str, Any],
    path: Path,
    *,
    transitioned_at: datetime,
) -> None:
    """Validate and durably bind the controller-created evidence exactly once."""
    if pending.get("schema_version") != PENDING_SCHEMA or pending.get("phase") != "AWAITING_EVIDENCE":
        raise core.CollectorError("E_CONTROLLER_STATE", "Controller evidence can only bind the active runtime-v2 run.", safety=True)
    deadline = core.parse_datetime(pending["deadline_at"], "pending.deadline_at")
    if transitioned_at > deadline:
        raise core.CollectorError(
            "E_OVERALL_DEADLINE",
            "Controller evidence cannot be bound after the total deadline.",
            safety=True,
        )
    _, payload, _, _ = _validate_evidence(config, pending, path)
    pending["evidence_identity"] = {"bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()}
    pending["phase"] = "EVIDENCE_BOUND"
    pending["last_transition_at"] = core.canonical_datetime(transitioned_at)
    _write_pending(config, pending)
    if _load_pending(config) != pending:
        raise core.CollectorError("E_CONTROLLER_STATE", "Controller evidence binding readback mismatch.", safety=True)


def _formal_tokens(event: Mapping[str, Any], config: core.CollectorConfig) -> list[str]:
    stable = event.get("stable_id")
    tokens: list[str] = []
    legacy_image: re.Match[str] | None = None
    if isinstance(stable, str) and stable:
        if re.fullmatch(r"BV[0-9A-Za-z]{10}", stable, re.IGNORECASE):
            tokens.append(f"bvid:{stable.lower()}")
        elif re.fullmatch(r"[0-9]{1,32}", stable):
            tokens.append(f"opus:{stable}")
        else:
            legacy_image = re.fullmatch(r"([0-9]{1,32}):(image|cover):([1-9][0-9]*)", stable)
        if legacy_image is not None and (
            event.get("item_type") == legacy_image.group(2)
            and event.get("source_parent_stable_id") == legacy_image.group(1)
        ):
            # Exact legacy image rows are independently deduped by their
            # canonical image URL.  Do not merge them into the parent opus
            # component because one parent can legitimately own many URLs.
            pass
        elif not tokens:
            raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal stable_id is invalid.", safety=True)
    source = event.get("source_url")
    if isinstance(source, str):
        canonical = core.validate_url(source, "formal.source_url", config.allowed_source_hosts)
        if legacy_image is not None:
            parsed = urlsplit(source)
            if (
                parsed.scheme != "https" or parsed.hostname not in {"i0.hdslb.com", "i1.hdslb.com", "i2.hdslb.com"}
                or parsed.query or parsed.fragment or not parsed.path.startswith("/bfs/")
            ):
                raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal image URL identity conflicts.", safety=True)
            canonical = source
        else:
            canonical = core.validate_url(source, "formal.source_url", config.allowed_source_hosts)
        tokens.append(f"url:{canonical}")
        path_parts = PurePosixPath(canonical.split("?", 1)[0].split("#", 1)[0]).parts
        if len(path_parts) >= 3 and path_parts[-2] == "opus" and path_parts[-1].isdecimal():
@@ -2302,10 +2284,14 @@
            evidence_hash=None, coverage=None, input_count=0, new_count=0,
            created=[], formal_changed=False,
        )
    if pending.get("phase") != "EVIDENCE_BOUND":
        raise core.CollectorError(
            "E_CONTROLLER_REQUIRED",
            "Schema 3 evidence is accepted only after the trusted refresh-run controller binds it.",
            safety=True,
        )
    evidence, payload, observation, identity_match = _validate_evidence(config, pending, evidence_path)
    evidence_hash = hashlib.sha256(payload).hexdigest()
    if pending.get("phase") != "EVIDENCE_BOUND":
        raise core.CollectorError("E_CONTROLLER_REQUIRED", "Evidence must be durably bound by the trusted controller.", safety=True)
    if pending.get("evidence_identity") != {"bytes": len(payload), "sha256": evidence_hash}:
        raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Bound evidence identity drifted.", safety=True)
    action_outcome = evidence["runtime_observation"]["refresh_action_outcome"]