From 873dca205129f123d5ea9dd768bfa1c2e5452830 Mon Sep 17 00:00:00 2001
From: MB-X Bilibili Pipeline <mbx-bili-pipeline@localhost>
Date: Sat, 05 Sep 2026 22:27:58 +0800
Subject: [PATCH] chore(project-info): archive Bilibili dynamic 66e47c0ffa2d
---
dev/project-dev/bili_dynamic_refresh.py | 150 ++++++++++++++++++++++---------------------------
1 files changed, 68 insertions(+), 82 deletions(-)
diff --git a/dev/project-dev/bili_dynamic_refresh.py b/dev/project-dev/bili_dynamic_refresh.py
index d7aeb3c..ab94822 100644
--- a/dev/project-dev/bili_dynamic_refresh.py
+++ b/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"]
--
Gitblit v1.9.3