"""Authenticated download worker security boundary.
|
|
This module is imported only by the worker mode. It owns yt-dlp, local-only
|
FFmpeg operations, and the frozen ``accept-browser-file`` bridge invocation.
|
Signed media URLs and cookies never leave this process memory.
|
"""
|
|
from __future__ import annotations
|
|
import ctypes
|
import hashlib
|
import importlib.util
|
import io
|
import json
|
import math
|
import os
|
import re
|
import shutil
|
import stat
|
import subprocess
|
import sys
|
import time
|
import uuid
|
from collections.abc import Mapping
|
from contextlib import contextmanager
|
from dataclasses import dataclass
|
from pathlib import Path
|
from typing import Any, Callable, Iterable, Sequence
|
from urllib.parse import urlparse
|
|
from .constants import (
|
BRIDGE_TIMEOUT_SECONDS,
|
EXTENSION_BUILD,
|
EXTRACTOR_RETRIES,
|
FILE_ACCESS_RETRIES,
|
FRAGMENT_RETRIES,
|
HTTP_RETRIES,
|
RELOAD_GENERATION,
|
SOCKET_TIMEOUT_SECONDS,
|
YTDLP_MODULE_SHA256,
|
YTDLP_VERSION,
|
duration_tolerance_ms,
|
validate_bvid,
|
validate_creator_uid,
|
)
|
from .formal_legacy_identity_manifest import (
|
FORMAL_LEGACY_CREATOR_UID,
|
FORMAL_LEGACY_INTEGER_UID_ROWS,
|
FORMAL_LEGACY_ROWS,
|
FORMAL_PREFIX_BYTES,
|
FORMAL_PREFIX_LINES,
|
FORMAL_PREFIX_SHA256,
|
)
|
from .protocol import (
|
ProtocolError, strict_json_loads, validate_media_complete_identity, validate_start,
|
)
|
|
FROZEN_BRIDGE_SHA256 = "00F11DAF8387160DB863C89F0B33AB8480422233FF42199189222C989C7ED07E"
|
SUBPROCESS_POLICY_ERROR_CODES = frozenset({
|
"E_SUBPROCESS_POLICY_ARGUMENTS",
|
"E_SUBPROCESS_POLICY_ENVIRONMENT",
|
"E_SUBPROCESS_POLICY_EVENT_SHAPE",
|
"E_SUBPROCESS_POLICY_EXECUTABLE",
|
"E_SUBPROCESS_POLICY_LOCAL_PATH",
|
"E_SUBPROCESS_POLICY_SECRET",
|
})
|
_BRIDGE_MAPPING_KEYS = frozenset({
|
"schema_version", "bvid", "source", "published_at", "title", "local_file",
|
"bytes", "sha256", "duration_seconds", "remote_duration_seconds",
|
"local_duration_seconds", "duration_delta_seconds", "duration_tolerance_seconds",
|
"format_name", "video_codec", "audio_codec", "completed_at", "acquisition_mode",
|
"handoff_source_sha256",
|
})
|
_BRIDGE_ITEM_KEYS = _BRIDGE_MAPPING_KEYS | {"status"}
|
_BRIDGE_ITEM_WARNING_KEYS = _BRIDGE_ITEM_KEYS | {"warning"}
|
_LOWER_SHA256_RE = re.compile(r"[0-9a-f]{64}")
|
_BRIDGE_CLEANUP_WARNING_RE = re.compile(
|
r"staging cleanup requires attention: [A-Za-z][A-Za-z0-9_]{0,63}"
|
)
|
_UTC_ISO_RE = re.compile(
|
r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?\+00:00"
|
)
|
|
|
class WorkerError(Exception):
|
def __init__(self, code: str, diagnostic: dict[str, object] | None = None) -> None:
|
super().__init__(code)
|
self.code = code
|
self.diagnostic = diagnostic
|
|
|
class CancelRequested(BaseException):
|
pass
|
|
|
class NullLogger:
|
def debug(self, _: object) -> None:
|
return None
|
|
def info(self, _: object) -> None:
|
return None
|
|
def warning(self, _: object) -> None:
|
return None
|
|
def error(self, _: object) -> None:
|
return None
|
|
|
def sha256_file(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as source:
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest().upper()
|
|
|
def verify_frozen_ytdlp() -> Path:
|
"""Verify the source package without importing it."""
|
spec = importlib.util.find_spec("yt_dlp")
|
if spec is None or spec.origin is None:
|
raise WorkerError("E_YTDLP_FROZEN")
|
root = Path(spec.origin).resolve().parent
|
for relative, expected in YTDLP_MODULE_SHA256.items():
|
path = root / Path(relative)
|
if not path.is_file() or sha256_file(path) != expected:
|
raise WorkerError("E_YTDLP_FROZEN")
|
version_path = root / "version.py"
|
version_scope: dict[str, Any] = {}
|
exec(compile(version_path.read_bytes(), str(version_path), "exec"), version_scope)
|
if version_scope.get("__version__") != YTDLP_VERSION:
|
raise WorkerError("E_YTDLP_FROZEN")
|
return root
|
|
|
def bootstrap_ytdlp() -> tuple[Any, Any]:
|
"""Disable every plugin source before the first yt-dlp import."""
|
os.environ["YTDLP_NO_PLUGINS"] = "1"
|
verify_frozen_ytdlp()
|
import yt_dlp # noqa: PLC0415 - deliberately after the environment gate
|
import yt_dlp.globals as yt_globals # noqa: PLC0415
|
import yt_dlp.plugins as yt_plugins # noqa: PLC0415
|
|
yt_globals.plugin_dirs.value = []
|
yt_plugins.load_all_plugins()
|
plugin_overrides = dict(yt_globals.plugin_ies_overrides.value)
|
if not (
|
yt_globals.plugin_dirs.value == []
|
and yt_plugins.directories() == []
|
and yt_globals.plugin_ies.value == {}
|
and yt_globals.plugin_pps.value == {}
|
and plugin_overrides == {}
|
):
|
raise WorkerError("E_PLUGIN_BOUNDARY")
|
return yt_dlp, yt_globals
|
|
|
@dataclass(frozen=True)
|
class HostConfig:
|
creator_allowlist: frozenset[str]
|
ffmpeg: Path
|
ffprobe: Path
|
bridge_python: Path
|
bridge_script: Path
|
yt_dlp_executable: Path
|
destination: Path
|
queue_lock_path: Path | None = None
|
formal_manifest_path: Path | None = None
|
processing_handoff_path: Path | None = None
|
creator_name: str = ""
|
|
@staticmethod
|
def _safe_absolute_file(value: Any, expected_hash: Any) -> Path:
|
if not isinstance(value, str) or not isinstance(expected_hash, str):
|
raise WorkerError("E_CONFIG")
|
path = Path(value)
|
if not path.is_absolute() or str(path).startswith("\\\\"):
|
raise WorkerError("E_CONFIG")
|
resolved = path.resolve(strict=True)
|
if not resolved.is_file() or resolved.is_symlink():
|
raise WorkerError("E_CONFIG")
|
if sha256_file(resolved) != expected_hash.upper():
|
raise WorkerError("E_CONFIG_HASH")
|
return resolved
|
|
@classmethod
|
def load(cls, path: Path) -> "HostConfig":
|
try:
|
if path.is_symlink():
|
raise WorkerError("E_CONFIG")
|
raw = strict_json_loads(path.read_bytes())
|
except (OSError, UnicodeError, json.JSONDecodeError, ProtocolError) as exc:
|
raise WorkerError("E_CONFIG") from exc
|
expected = {
|
"schema",
|
"creator_allowlist",
|
"queue_path",
|
"queue_state_path",
|
"queue_lock_path",
|
"reload_state_path",
|
"reload_generation",
|
"required_extension_build",
|
"ffmpeg",
|
"ffmpeg_sha256",
|
"ffprobe",
|
"ffprobe_sha256",
|
"bridge_python",
|
"bridge_python_sha256",
|
"bridge_script",
|
"bridge_script_sha256",
|
"yt_dlp_executable",
|
"yt_dlp_executable_sha256",
|
"destination",
|
"formal_manifest_path",
|
"processing_handoff_path",
|
"creator_name",
|
}
|
if not isinstance(raw, dict) or set(raw) != expected:
|
raise WorkerError("E_CONFIG")
|
if raw["schema"] != 2:
|
raise WorkerError("E_CONFIG")
|
if raw["required_extension_build"] != EXTENSION_BUILD or raw["reload_generation"] != RELOAD_GENERATION:
|
raise WorkerError("E_CONFIG")
|
creators = raw["creator_allowlist"]
|
try:
|
if (
|
not isinstance(creators, list) or not creators or len(creators) > 64
|
or [validate_creator_uid(item) for item in creators] != sorted(set(creators))
|
):
|
raise WorkerError("E_CONFIG")
|
except ValueError as exc:
|
raise WorkerError("E_CONFIG") from exc
|
for name in (
|
"queue_path", "queue_state_path", "queue_lock_path", "reload_state_path",
|
"formal_manifest_path", "processing_handoff_path",
|
):
|
value = raw[name]
|
if not isinstance(value, str):
|
raise WorkerError("E_CONFIG")
|
creator_name = raw["creator_name"]
|
if (
|
not isinstance(creator_name, str) or not creator_name.strip()
|
or len(creator_name.encode("utf-8")) > 240
|
or any(ord(ch) < 32 or ord(ch) == 127 for ch in creator_name)
|
):
|
raise WorkerError("E_CONFIG")
|
queue_lock_path = Path(raw["queue_lock_path"])
|
formal_manifest_path = Path(raw["formal_manifest_path"])
|
processing_handoff_path = Path(raw["processing_handoff_path"])
|
governed_paths: list[Path] = []
|
for governed_path, must_exist in (
|
(queue_lock_path, False),
|
(formal_manifest_path, True),
|
(processing_handoff_path, False),
|
):
|
if not governed_path.is_absolute() or str(governed_path).startswith("\\\\"):
|
raise WorkerError("E_CONFIG")
|
try:
|
lexical_parent = governed_path.parent
|
parent_stat = lexical_parent.lstat()
|
resolved_parent = lexical_parent.resolve(strict=True)
|
if (
|
not stat.S_ISDIR(parent_stat.st_mode)
|
or lexical_parent.is_symlink()
|
or _is_reparse(lexical_parent)
|
or resolved_parent != lexical_parent
|
):
|
raise WorkerError("E_CONFIG")
|
resolved_path = resolved_parent / governed_path.name
|
try:
|
path_stat = governed_path.lstat()
|
except FileNotFoundError:
|
if must_exist:
|
raise WorkerError("E_CONFIG")
|
else:
|
if (
|
not stat.S_ISREG(path_stat.st_mode)
|
or governed_path.is_symlink()
|
or _is_reparse(governed_path)
|
or governed_path.resolve(strict=True) != resolved_path
|
):
|
raise WorkerError("E_CONFIG")
|
except WorkerError:
|
raise
|
except OSError as exc:
|
raise WorkerError("E_CONFIG") from exc
|
governed_paths.append(resolved_path)
|
if len(set(governed_paths)) != len(governed_paths):
|
raise WorkerError("E_CONFIG")
|
queue_lock_path, formal_manifest_path, processing_handoff_path = governed_paths
|
bridge_script = cls._safe_absolute_file(raw["bridge_script"], raw["bridge_script_sha256"])
|
if raw["bridge_script_sha256"].upper() != FROZEN_BRIDGE_SHA256:
|
raise WorkerError("E_CONFIG_HASH")
|
destination = Path(raw["destination"])
|
if not destination.is_absolute() or str(destination).startswith("\\\\"):
|
raise WorkerError("E_CONFIG")
|
destination = destination.resolve(strict=True)
|
if not destination.is_dir() or destination.is_symlink():
|
raise WorkerError("E_CONFIG")
|
return cls(
|
creator_allowlist=frozenset(creators),
|
ffmpeg=cls._safe_absolute_file(raw["ffmpeg"], raw["ffmpeg_sha256"]),
|
ffprobe=cls._safe_absolute_file(raw["ffprobe"], raw["ffprobe_sha256"]),
|
bridge_python=cls._safe_absolute_file(
|
raw["bridge_python"], raw["bridge_python_sha256"]
|
),
|
bridge_script=bridge_script,
|
yt_dlp_executable=cls._safe_absolute_file(
|
raw["yt_dlp_executable"], raw["yt_dlp_executable_sha256"]
|
),
|
destination=destination,
|
queue_lock_path=queue_lock_path,
|
formal_manifest_path=formal_manifest_path,
|
processing_handoff_path=processing_handoff_path,
|
creator_name=creator_name,
|
)
|
|
|
def _is_reparse(path: Path) -> bool:
|
try:
|
attributes = path.lstat().st_file_attributes
|
except AttributeError:
|
return path.is_symlink()
|
return bool(attributes & 0x400)
|
|
|
def _reject_reparse_path(path: Path, stop: Path | None = None) -> None:
|
current = path
|
stop_value = stop.resolve(strict=False) if stop is not None else None
|
while True:
|
if current.exists() and _is_reparse(current):
|
raise WorkerError("E_STAGE")
|
if (stop_value is not None and current.resolve(strict=False) == stop_value) or current.parent == current:
|
break
|
current = current.parent
|
|
|
def validated_local_app_data() -> Path:
|
"""Return the one non-secret environment path retained by the worker."""
|
value = os.environ.get("LOCALAPPDATA")
|
if not value:
|
raise WorkerError("E_STAGE")
|
candidate = Path(value)
|
if not candidate.is_absolute() or str(candidate).startswith("\\\\"):
|
raise WorkerError("E_STAGE")
|
_reject_reparse_path(candidate)
|
try:
|
resolved = candidate.resolve(strict=True)
|
except OSError as exc:
|
raise WorkerError("E_STAGE") from exc
|
if not resolved.is_dir() or _is_reparse(resolved):
|
raise WorkerError("E_STAGE")
|
return resolved
|
|
|
def fixed_stage_root(bvid: str) -> Path:
|
local_app_data = validated_local_app_data()
|
try:
|
bvid = validate_bvid(bvid)
|
except ValueError as exc:
|
raise WorkerError("E_STAGE") from exc
|
logical_root = (
|
local_app_data
|
/ "project-info"
|
/ "bili-auth-ingress"
|
/ bvid
|
)
|
_reject_reparse_path(logical_root, local_app_data)
|
resolved = logical_root.resolve(strict=False)
|
_ensure_within(resolved, local_app_data)
|
return resolved
|
|
|
def _ensure_within(path: Path, root: Path) -> Path:
|
resolved = path.resolve(strict=False)
|
try:
|
resolved.relative_to(root.resolve(strict=False))
|
except ValueError as exc:
|
raise WorkerError("E_STAGE") from exc
|
return resolved
|
|
|
def _reject_reparse_chain(path: Path, stop: Path) -> None:
|
_reject_reparse_path(path, stop)
|
|
|
def cleanup_stale_runs(root: Path, *, boundary: Path | None = None) -> None:
|
"""Remove only uncommitted run-* directories below the fixed stage root."""
|
allowed_root = root.resolve() if boundary is None else boundary.resolve()
|
_ensure_within(root, allowed_root)
|
if not root.exists():
|
return
|
_reject_reparse_chain(root, allowed_root)
|
for child in root.iterdir():
|
if not child.name.startswith("run-") or not child.is_dir() or child.is_symlink():
|
raise WorkerError("E_STAGE")
|
_ensure_within(child, root)
|
shutil.rmtree(child)
|
|
|
def create_run_directory(root: Path) -> Path:
|
stage_root = root.resolve()
|
_ensure_within(stage_root, stage_root)
|
stage_root.mkdir(parents=True, exist_ok=True)
|
_reject_reparse_chain(stage_root, stage_root)
|
for _ in range(8):
|
candidate = stage_root / f"run-{uuid.uuid4().hex}"
|
try:
|
candidate.mkdir(exist_ok=False)
|
return candidate
|
except FileExistsError:
|
continue
|
raise WorkerError("E_STAGE")
|
|
|
def prepare_run_directory(root: Path) -> Path:
|
"""Clean stale runs and create the secret-free task lease."""
|
stage_root = root.resolve()
|
cleanup_stale_runs(stage_root, boundary=stage_root)
|
return create_run_directory(stage_root)
|
|
|
def cleanup_run_directory(run_directory: Path, stage_root: Path) -> None:
|
_ensure_within(run_directory, stage_root)
|
if run_directory.exists():
|
if not run_directory.is_dir() or _is_reparse(run_directory):
|
raise WorkerError("E_STAGE")
|
shutil.rmtree(run_directory)
|
if stage_root.exists() and not any(stage_root.iterdir()):
|
stage_root.rmdir()
|
|
|
def build_cookie_stream(start: dict[str, Any]) -> io.StringIO:
|
"""Convert validated Chrome records to an in-memory Netscape jar."""
|
validate_start(start)
|
stream = io.StringIO(newline="\n")
|
stream.write("# Netscape HTTP Cookie File\n")
|
for cookie in start["cookies"]:
|
domain = cookie["domain"]
|
if cookie["http_only"]:
|
domain = f"#HttpOnly_{domain}"
|
fields = (
|
domain,
|
"FALSE" if cookie["host_only"] else "TRUE",
|
cookie["path"],
|
"TRUE" if cookie["secure"] else "FALSE",
|
"0" if cookie["session"] else str(cookie["expiration_unix"]),
|
cookie["name"],
|
cookie["value"],
|
)
|
stream.write("\t".join(fields) + "\n")
|
stream.seek(0)
|
return stream
|
|
|
def close_cookie_stream(stream: io.StringIO | None) -> bool:
|
if stream is None:
|
return True
|
try:
|
stream.seek(0)
|
stream.truncate(0)
|
finally:
|
stream.close()
|
return stream.closed
|
|
|
def _finite_number(value: Any) -> float:
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
raise WorkerError("E_METADATA")
|
converted = float(value)
|
if not math.isfinite(converted):
|
raise WorkerError("E_METADATA")
|
return converted
|
|
|
def validate_processed_info(info: Any, job: dict[str, Any]) -> dict[str, Any]:
|
if not isinstance(info, dict) or info.get("id") != job["bvid"]:
|
raise WorkerError("E_METADATA")
|
if info.get("entries") not in (None, []) or info.get("_type") not in (None, "video"):
|
raise WorkerError("E_MULTI_PART")
|
if info.get("playlist_count") not in (None, 1) or info.get("playlist_index") not in (None, 1):
|
raise WorkerError("E_MULTI_PART")
|
if info.get("is_live") is True or info.get("live_status") not in (None, "not_live"):
|
raise WorkerError("E_LIVE")
|
if info.get("has_drm") is True:
|
raise WorkerError("E_DRM")
|
if info.get("availability") not in (None, "public", "unlisted"):
|
raise WorkerError("E_ENTITLEMENT")
|
owner_ids = {str(value) for value in (info.get("uploader_id"), info.get("channel_id")) if value is not None}
|
if job["creator_uid"] not in owner_ids:
|
raise WorkerError("E_OWNER")
|
duration_ms = round(_finite_number(info.get("duration")) * 1000)
|
if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
|
raise WorkerError("E_DURATION")
|
formats = info.get("formats")
|
if not isinstance(formats, list) or not formats:
|
raise WorkerError("E_FORMAT")
|
return info
|
|
|
def _format_leaves(download_info: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
|
requested = download_info.get("requested_formats")
|
if requested is None:
|
leaves = [download_info]
|
single = True
|
else:
|
if not isinstance(requested, list) or len(requested) != 2:
|
raise WorkerError("E_FORMAT")
|
if not all(isinstance(item, dict) for item in requested):
|
raise WorkerError("E_FORMAT")
|
leaves = requested
|
single = False
|
return leaves, single
|
|
|
def validate_download_info(
|
download_info: Any,
|
params: dict[str, Any],
|
job: dict[str, Any],
|
*,
|
downloader_resolver: Callable[..., Any] | None = None,
|
) -> tuple[list[dict[str, Any]], bool]:
|
if not isinstance(download_info, dict) or download_info.get("id") != job["bvid"]:
|
raise WorkerError("E_FORMAT")
|
leaves, single = _format_leaves(download_info)
|
if single:
|
if download_info.get("vcodec") in (None, "none") or download_info.get("acodec") in (None, "none"):
|
raise WorkerError("E_FORMAT")
|
else:
|
video_only = sum(
|
leaf.get("vcodec") not in (None, "none") and leaf.get("acodec") == "none"
|
for leaf in leaves
|
)
|
audio_only = sum(
|
leaf.get("vcodec") == "none" and leaf.get("acodec") not in (None, "none")
|
for leaf in leaves
|
)
|
if video_only != 1 or audio_only != 1:
|
raise WorkerError("E_FORMAT")
|
if downloader_resolver is None:
|
from yt_dlp.downloader import get_suitable_downloader # noqa: PLC0415
|
|
downloader_resolver = get_suitable_downloader
|
for leaf in leaves:
|
if leaf.get("has_drm") is True:
|
raise WorkerError("E_DRM")
|
url = leaf.get("url")
|
protocol = leaf.get("protocol")
|
if not isinstance(url, str) or urlparse(url).scheme != "https":
|
raise WorkerError("E_FORMAT")
|
if protocol not in {"https", "http_dash_segments"}:
|
raise WorkerError("E_DOWNLOADER")
|
downloader = downloader_resolver(leaf, params)
|
if getattr(downloader, "__name__", "") not in {"HttpFD", "DashSegmentsFD"}:
|
raise WorkerError("E_DOWNLOADER")
|
return leaves, single
|
|
|
def prepare_download_info(
|
ydl: Any,
|
job: dict[str, Any],
|
*,
|
downloader_resolver: Callable[..., Any] | None = None,
|
) -> tuple[dict[str, Any], bool, tuple[str, ...]]:
|
extract_count = 0
|
original_extract = ydl.extract_info
|
|
def one_extract(*args: Any, **kwargs: Any) -> Any:
|
nonlocal extract_count
|
extract_count += 1
|
if extract_count != 1:
|
raise WorkerError("E_SECOND_EXTRACT")
|
return original_extract(*args, **kwargs)
|
|
ydl.extract_info = one_extract
|
processed = ydl.extract_info(job["canonical_url"], download=False, process=True)
|
validate_processed_info(processed, job)
|
selector = ydl.build_format_selector("bestvideo+bestaudio/best")
|
selected = list(ydl._select_formats(ydl._get_formats(processed), selector))
|
if len(selected) != 1:
|
raise WorkerError("E_FORMAT")
|
download_info = ydl._copy_infodict(processed)
|
download_info.update(selected[0])
|
leaves, single = validate_download_info(
|
download_info,
|
ydl.params,
|
job,
|
downloader_resolver=downloader_resolver,
|
)
|
signed_urls = tuple(str(leaf["url"]) for leaf in leaves)
|
|
def reject_second_extract(*_: Any, **__: Any) -> Any:
|
raise WorkerError("E_SECOND_EXTRACT")
|
|
ydl.extract_info = reject_second_extract
|
if extract_count != 1:
|
raise WorkerError("E_SECOND_EXTRACT")
|
return download_info, single, signed_urls
|
|
|
class SubprocessPolicy:
|
"""Pre-CreateProcess audit for local-only child command lines."""
|
|
_FORBIDDEN = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
|
_FFMPEG_FLAGS = frozenset({"-nostdin", "-y"})
|
_FFMPEG_SCALAR_OPTIONS = frozenset(
|
{"-v", "-loglevel", "-map", "-c", "-map_metadata", "-f", "-movflags"}
|
)
|
_FFMPEG_REPEATABLE_OPTIONS = frozenset({"-map"})
|
_FFPROBE_FLAGS = frozenset({"-hide_banner", "-show_format", "-show_streams"})
|
_FFPROBE_SCALAR_OPTIONS = frozenset({"-v", "-show_entries", "-of", "-print_format"})
|
_WINDOWS_RESERVED_NAMES = frozenset(
|
{"CON", "PRN", "AUX", "NUL", *(f"COM{index}" for index in range(1, 10)),
|
*(f"LPT{index}" for index in range(1, 10))}
|
)
|
_MAX_ARGUMENTS = 256
|
_MAX_COMMAND_LINE = 32_767
|
_MAX_ENVIRONMENT_ITEMS = 256
|
_MAX_LOCAL_FILE_OPERANDS = 16
|
|
def __init__(
|
self,
|
run_root: Path,
|
executables: Iterable[Path],
|
secrets: Iterable[str] = (),
|
) -> None:
|
self.run_root = run_root.resolve()
|
self.executables = {os.path.normcase(str(item.resolve())) for item in executables}
|
self.secrets = {item.casefold() for item in secrets if item}
|
|
@staticmethod
|
def _fail(code: str) -> None:
|
if code not in SUBPROCESS_POLICY_ERROR_CODES:
|
code = "E_SUBPROCESS_POLICY_EVENT_SHAPE"
|
raise WorkerError(code)
|
|
@classmethod
|
def _windows_arguments(cls, command_line: Any) -> list[str]:
|
if (
|
not isinstance(command_line, str)
|
or not command_line
|
or len(command_line) > cls._MAX_COMMAND_LINE
|
or "\x00" in command_line
|
):
|
cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
argc = ctypes.c_int()
|
command_line_to_argv = ctypes.windll.shell32.CommandLineToArgvW
|
command_line_to_argv.argtypes = (ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_int))
|
command_line_to_argv.restype = ctypes.POINTER(ctypes.c_wchar_p)
|
pointer = command_line_to_argv(command_line, ctypes.byref(argc))
|
if not pointer:
|
cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
try:
|
result = [pointer[index] for index in range(argc.value)]
|
finally:
|
local_free = ctypes.windll.kernel32.LocalFree
|
local_free.argtypes = (ctypes.c_void_p,)
|
local_free.restype = ctypes.c_void_p
|
local_free(ctypes.cast(pointer, ctypes.c_void_p))
|
if (
|
not result
|
or len(result) > cls._MAX_ARGUMENTS
|
or any(not isinstance(item, str) or "\x00" in item for item in result)
|
or subprocess.list2cmdline(result) != command_line
|
):
|
cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
return result
|
|
@classmethod
|
def _portable_arguments(cls, raw: Any) -> list[str]:
|
if not isinstance(raw, (list, tuple)) or not raw or len(raw) > cls._MAX_ARGUMENTS:
|
cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
if any(type(item) is not str or "\x00" in item for item in raw):
|
cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
return list(raw)
|
|
@staticmethod
|
def _absolute_executable(value: Any) -> tuple[str, Path]:
|
if not isinstance(value, (str, os.PathLike)):
|
SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
filesystem_value = os.fspath(value)
|
if not isinstance(filesystem_value, str) or not filesystem_value or "\x00" in filesystem_value:
|
SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
path = Path(filesystem_value)
|
if not path.is_absolute() or str(path).startswith("\\\\"):
|
SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
|
try:
|
resolved = path.resolve(strict=True)
|
except OSError as exc:
|
raise WorkerError("E_SUBPROCESS_POLICY_EXECUTABLE") from exc
|
if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
|
SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
|
return os.path.normcase(str(resolved)), resolved
|
|
def _validate_environment(self, environment: Any) -> None:
|
if environment is None:
|
return
|
if not isinstance(environment, Mapping) or len(environment) > self._MAX_ENVIRONMENT_ITEMS:
|
self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
|
encoded: list[str] = []
|
for key, value in environment.items():
|
if (
|
type(key) is not str
|
or type(value) is not str
|
or not key
|
or "\x00" in key
|
or "\x00" in value
|
):
|
self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
|
encoded.append(f"{key}={value}")
|
folded = "\x00".join(encoded).casefold()
|
if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
|
self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
|
|
def check(self, event: str, arguments: tuple[Any, ...]) -> None:
|
if event != "subprocess.Popen":
|
return
|
if not isinstance(arguments, tuple) or len(arguments) != 4:
|
self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
executable, raw_arguments, cwd, environment = arguments
|
if cwd is not None:
|
self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
|
text_args = (
|
self._windows_arguments(raw_arguments)
|
if os.name == "nt"
|
else self._portable_arguments(raw_arguments)
|
)
|
argv_executable_key, _ = self._absolute_executable(text_args[0])
|
if executable is None:
|
executable_key = argv_executable_key
|
else:
|
executable_key, _ = self._absolute_executable(executable)
|
if executable_key != argv_executable_key:
|
self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
|
if executable_key not in self.executables:
|
self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
|
folded = "\x00".join(text_args).casefold()
|
if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
|
self._fail("E_SUBPROCESS_POLICY_SECRET")
|
self._validate_environment(environment)
|
executable_name = Path(executable_key).name.casefold()
|
if executable_name == "ffmpeg.exe":
|
self._validate_ffmpeg_arguments(text_args)
|
elif executable_name == "ffprobe.exe":
|
self._validate_ffprobe_arguments(text_args)
|
|
@staticmethod
|
def _valid_ffmpeg_scalar(option: str, value: str) -> bool:
|
if option == "-v":
|
return value == "error"
|
if option == "-loglevel":
|
return value == "repeat+info"
|
if option == "-map":
|
return re.fullmatch(r"\d+(?::[av](?::\d+)?)?", value) is not None
|
if option == "-c":
|
return value == "copy"
|
if option == "-map_metadata":
|
return value == "-1"
|
if option == "-f":
|
return value == "matroska"
|
if option == "-movflags":
|
return value == "+faststart"
|
return False
|
|
@staticmethod
|
def _valid_ffprobe_scalar(option: str, value: str) -> bool:
|
if option == "-v":
|
return value == "error"
|
if option == "-show_entries":
|
return value == "format=format_name,duration:stream=codec_type"
|
if option in {"-of", "-print_format"}:
|
return value == "json"
|
return False
|
|
def _validate_ffmpeg_arguments(self, arguments: Sequence[str]) -> None:
|
if tuple(arguments[1:]) == ("-bsfs",):
|
return
|
index = 1
|
input_count = 0
|
file_operand_count = 0
|
output_seen = False
|
seen_options: set[str] = set()
|
while index < len(arguments):
|
option = arguments[index]
|
if option in self._FFMPEG_FLAGS:
|
if option in seen_options:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
seen_options.add(option)
|
index += 1
|
continue
|
if option in {"-i", "-attach"}:
|
if index + 1 >= len(arguments) or arguments[index + 1].startswith("-"):
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
self._local_path(arguments[index + 1], must_exist=True)
|
file_operand_count += 1
|
input_count += option == "-i"
|
if file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
index += 2
|
continue
|
if option in self._FFMPEG_SCALAR_OPTIONS:
|
if index + 1 >= len(arguments):
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
value = arguments[index + 1]
|
if not self._valid_ffmpeg_scalar(option, value):
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
if option not in self._FFMPEG_REPEATABLE_OPTIONS:
|
if option in seen_options:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
seen_options.add(option)
|
index += 2
|
continue
|
if re.fullmatch(r"-bsf:a:\d+", option):
|
if index + 1 >= len(arguments) or arguments[index + 1] != "aac_adtstoasc":
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
if option in seen_options:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
seen_options.add(option)
|
index += 2
|
continue
|
if option.startswith("-") or output_seen or index != len(arguments) - 1:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
self._local_path(option, must_exist=False)
|
output_seen = True
|
file_operand_count += 1
|
index += 1
|
if input_count < 1 or not output_seen or file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
|
def _validate_ffprobe_arguments(self, arguments: Sequence[str]) -> None:
|
if tuple(arguments[1:]) == ("-bsfs",):
|
return
|
index = 1
|
input_seen = False
|
seen_options: set[str] = set()
|
while index < len(arguments):
|
option = arguments[index]
|
if option in self._FFPROBE_FLAGS:
|
if option in seen_options:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
seen_options.add(option)
|
index += 1
|
continue
|
if option in self._FFPROBE_SCALAR_OPTIONS:
|
if index + 1 >= len(arguments) or not self._valid_ffprobe_scalar(
|
option, arguments[index + 1]
|
):
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
if option in seen_options:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
seen_options.add(option)
|
index += 2
|
continue
|
if option.startswith("-") or input_seen or index != len(arguments) - 1:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
self._local_path(option, must_exist=True)
|
input_seen = True
|
index += 1
|
if not input_seen:
|
self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
|
|
def _local_path(self, value: str, *, must_exist: bool) -> Path:
|
if value.startswith("file:"):
|
value = value[5:]
|
path = Path(value)
|
if (
|
not value
|
or not path.is_absolute()
|
or str(path).startswith("\\\\")
|
or ":" in value[2:]
|
or any(part == ".." or part.endswith((" ", ".")) for part in path.parts)
|
or path.name.split(".", 1)[0].upper() in self._WINDOWS_RESERVED_NAMES
|
):
|
self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
|
try:
|
path.relative_to(self.run_root)
|
_reject_reparse_chain(path, self.run_root)
|
if must_exist:
|
resolved = path.resolve(strict=True)
|
resolved.relative_to(self.run_root)
|
if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
|
self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
|
else:
|
if path.exists() or path.is_symlink():
|
self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
|
parent = path.parent.resolve(strict=True)
|
parent.relative_to(self.run_root)
|
_reject_reparse_chain(path.parent, self.run_root)
|
if not parent.is_dir() or parent.is_symlink() or _is_reparse(parent):
|
self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
|
resolved = parent / path.name
|
except (OSError, ValueError, WorkerError) as exc:
|
raise WorkerError("E_SUBPROCESS_POLICY_LOCAL_PATH") from exc
|
return resolved
|
|
def install(self) -> None:
|
sys.addaudithook(self.check)
|
|
|
def sanitized_environment() -> dict[str, str]:
|
allowed = {"PATH", "PATHEXT", "SYSTEMROOT", "WINDIR", "TEMP", "TMP", "COMSPEC"}
|
result = {key: value for key, value in os.environ.items() if key.upper() in allowed}
|
result["LOCALAPPDATA"] = str(validated_local_app_data())
|
result["YTDLP_NO_PLUGINS"] = "1"
|
return result
|
|
|
def _run_local(
|
command: Sequence[str],
|
timeout: int,
|
*,
|
capture_stdout: bool = False,
|
) -> subprocess.CompletedProcess[bytes]:
|
return subprocess.run(
|
list(command),
|
stdin=subprocess.DEVNULL,
|
stdout=subprocess.PIPE if capture_stdout else subprocess.DEVNULL,
|
stderr=subprocess.DEVNULL,
|
check=False,
|
timeout=timeout,
|
env=sanitized_environment(),
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
)
|
|
|
def remux_single_to_mkv(ffmpeg: Path, source: Path, destination: Path) -> None:
|
if destination.exists():
|
raise WorkerError("E_COLLISION")
|
try:
|
result = _run_local(
|
[
|
str(ffmpeg),
|
"-nostdin",
|
"-v",
|
"error",
|
"-i",
|
str(source),
|
"-map",
|
"0:v:0",
|
"-map",
|
"0:a:0",
|
"-c",
|
"copy",
|
"-map_metadata",
|
"-1",
|
"-f",
|
"matroska",
|
str(destination),
|
],
|
timeout=BRIDGE_TIMEOUT_SECONDS,
|
)
|
except subprocess.TimeoutExpired as exc:
|
destination.unlink(missing_ok=True)
|
raise WorkerError("E_MERGE") from exc
|
if result.returncode != 0 or not destination.is_file() or destination.stat().st_size <= 0:
|
destination.unlink(missing_ok=True)
|
raise WorkerError("E_MERGE")
|
|
|
def merge_local_streams(ffmpeg: Path, video: Path, audio: Path, destination: Path) -> None:
|
if destination.exists():
|
raise WorkerError("E_COLLISION")
|
try:
|
result = _run_local(
|
[
|
str(ffmpeg),
|
"-nostdin",
|
"-v",
|
"error",
|
"-i",
|
str(video),
|
"-i",
|
str(audio),
|
"-map",
|
"0:v:0",
|
"-map",
|
"1:a:0",
|
"-c",
|
"copy",
|
"-map_metadata",
|
"-1",
|
"-f",
|
"matroska",
|
str(destination),
|
],
|
timeout=BRIDGE_TIMEOUT_SECONDS,
|
)
|
except subprocess.TimeoutExpired as exc:
|
destination.unlink(missing_ok=True)
|
raise WorkerError("E_MERGE") from exc
|
if result.returncode != 0 or not destination.is_file() or destination.stat().st_size <= 0:
|
destination.unlink(missing_ok=True)
|
raise WorkerError("E_MERGE")
|
|
|
def probe_mkv(ffprobe: Path, candidate: Path, job: dict[str, Any] | None = None) -> None:
|
try:
|
result = _run_local(
|
[
|
str(ffprobe),
|
"-v",
|
"error",
|
"-show_entries",
|
"format=format_name,duration:stream=codec_type",
|
"-of",
|
"json",
|
str(candidate),
|
],
|
timeout=120,
|
capture_stdout=True,
|
)
|
except subprocess.TimeoutExpired as exc:
|
raise WorkerError("E_MEDIA_VALIDATION") from exc
|
try:
|
payload = json.loads(result.stdout.decode("utf-8"))
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
raise WorkerError("E_MEDIA_VALIDATION") from exc
|
stream_types = {item.get("codec_type") for item in payload.get("streams", []) if isinstance(item, dict)}
|
format_name = payload.get("format", {}).get("format_name", "")
|
if result.returncode != 0 or {"video", "audio"} - stream_types or "matroska" not in format_name:
|
raise WorkerError("E_MEDIA_VALIDATION")
|
if job is not None:
|
raw_duration = payload.get("format", {}).get("duration")
|
if isinstance(raw_duration, str) and re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", raw_duration):
|
raw_duration = float(raw_duration)
|
duration_ms = round(_finite_number(raw_duration) * 1000)
|
if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
|
raise WorkerError("E_DURATION")
|
|
|
def validate_unique_candidate(run_directory: Path) -> Path:
|
forbidden_suffixes = {".part", ".tmp", ".crdownload", ".txt", ".json", ".url"}
|
files = [item for item in run_directory.iterdir() if item.is_file()]
|
if any(item.suffix.casefold() in forbidden_suffixes for item in files):
|
raise WorkerError("E_STAGE")
|
candidates = [item for item in files if item.suffix.casefold() == ".mkv"]
|
if len(candidates) != 1 or len(files) != 1:
|
raise WorkerError("E_STAGE")
|
return candidates[0]
|
|
|
def _ordinary_exact_file(path: Path, parent: Path) -> os.stat_result:
|
try:
|
if path.parent.resolve(strict=True) != parent.resolve(strict=True):
|
raise WorkerError("E_COLLISION")
|
_reject_reparse_path(path, parent)
|
value = path.lstat()
|
except OSError as exc:
|
raise WorkerError("E_COLLISION") from exc
|
if not stat.S_ISREG(value.st_mode) or _is_reparse(path):
|
raise WorkerError("E_COLLISION")
|
return value
|
|
|
def _stable_file_identity(value: os.stat_result) -> tuple[int, ...]:
|
"""Return cross-API file identity fields; content is bound separately by SHA."""
|
return (
|
value.st_dev,
|
value.st_ino,
|
value.st_mode,
|
value.st_nlink,
|
value.st_size,
|
)
|
|
|
def _stable_pair_test_seam(_: str) -> None:
|
"""Named no-op seams used only by production-shaped race regressions."""
|
return None
|
|
|
def _completion_test_seam(_: str) -> None:
|
"""Named no-op seams for durable completion transaction regressions."""
|
return None
|
|
|
def _canonical_json_line(value: Mapping[str, Any]) -> bytes:
|
return json.dumps(
|
dict(value), ensure_ascii=False, allow_nan=False, separators=(",", ":")
|
).encode("utf-8") + b"\n"
|
|
|
_FORMAL_PRIOR_REQUIRED_KEYS = frozenset({
|
"stable_id", "creator_uid", "source_url", "published_at", "item_type", "status",
|
})
|
_FORMAL_PRIOR_STATUS_RE = re.compile(r"VIDEO_[A-Z0-9_]{1,127}\Z")
|
_FORMAL_LEGACY_BY_LINE = {row[0]: row for row in FORMAL_LEGACY_ROWS}
|
_FORMAL_LEGACY_INTEGER_UID_BY_LINE = {
|
row[0]: row for row in FORMAL_LEGACY_INTEGER_UID_ROWS
|
}
|
|
|
def _formal_raw_lines(payload: bytes) -> list[bytes]:
|
complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
|
return complete.splitlines()
|
|
|
def _validate_formal_legacy_manifest() -> None:
|
if (
|
FORMAL_PREFIX_BYTES != 103_766
|
or FORMAL_PREFIX_LINES != 119
|
or not re.fullmatch(r"[A-F0-9]{64}", FORMAL_PREFIX_SHA256)
|
or not re.fullmatch(r"[1-9][0-9]{1,19}", FORMAL_LEGACY_CREATOR_UID)
|
or len(FORMAL_LEGACY_ROWS) != 26
|
or len(_FORMAL_LEGACY_BY_LINE) != len(FORMAL_LEGACY_ROWS)
|
or len(FORMAL_LEGACY_INTEGER_UID_ROWS) != 1
|
or len(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
|
!= len(FORMAL_LEGACY_INTEGER_UID_ROWS)
|
or set(_FORMAL_LEGACY_BY_LINE) & set(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
for row in FORMAL_LEGACY_ROWS:
|
if (
|
not isinstance(row, tuple) or len(row) != 8
|
or not isinstance(row[0], int) or not 1 <= row[0] <= FORMAL_PREFIX_LINES
|
or not isinstance(row[1], int) or row[1] < 2
|
or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
|
or not all(isinstance(value, str) and value for value in row[3:])
|
or row[6] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[7]) is None
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
for row in FORMAL_LEGACY_INTEGER_UID_ROWS:
|
if (
|
not isinstance(row, tuple) or len(row) != 9
|
or type(row[0]) is not int or not 1 <= row[0] <= FORMAL_PREFIX_LINES
|
or type(row[1]) is not int or row[1] < 2
|
or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
|
or type(row[3]) is not int or row[3] <= 0
|
or str(row[3]) != FORMAL_LEGACY_CREATOR_UID
|
or not all(isinstance(value, str) and value for value in row[4:])
|
or row[7] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[8]) is None
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
|
|
def _validate_legacy_formal_prior(
|
payload: bytes,
|
raw_lines: Sequence[bytes],
|
line_ordinal: int,
|
value: dict[str, Any],
|
config: HostConfig,
|
job: dict[str, Any],
|
) -> str:
|
_validate_formal_legacy_manifest()
|
if (
|
len(payload) < FORMAL_PREFIX_BYTES
|
or len(raw_lines) < FORMAL_PREFIX_LINES
|
or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper() != FORMAL_PREFIX_SHA256
|
or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
|
or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
|
or job["creator_uid"] != FORMAL_LEGACY_CREATOR_UID
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
expected = _FORMAL_LEGACY_BY_LINE.get(line_ordinal)
|
if expected is None:
|
raise WorkerError("E_COMPLETION_FORMAL")
|
raw_line = raw_lines[line_ordinal - 1]
|
if (
|
len(raw_line) != expected[1]
|
or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
|
or set(_FORMAL_PRIOR_REQUIRED_KEYS) - set(value) != {"creator_uid"}
|
or value.get("stable_id") != expected[3]
|
or value.get("source_url") != expected[4]
|
or value.get("published_at") != expected[5]
|
or value.get("item_type") != expected[6]
|
or value.get("status") != expected[7]
|
or value.get("schema_version") != 1
|
or value.get("creator") != config.creator_name
|
or expected[3] != job["bvid"]
|
or expected[4] != job["canonical_url"]
|
or expected[5] != job["published_at"]
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
return expected[7]
|
|
|
def _validate_current_formal_prior(
|
value: dict[str, Any], config: HostConfig, job: dict[str, Any]
|
) -> str:
|
if (
|
not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
|
or type(value.get("schema_version")) is not int
|
or value["schema_version"] != 1
|
or value.get("creator") != config.creator_name
|
or not isinstance(value.get("stable_id"), str)
|
or not isinstance(value.get("creator_uid"), str)
|
or not isinstance(value.get("source_url"), str)
|
or not isinstance(value.get("published_at"), str)
|
or not isinstance(value.get("item_type"), str)
|
or not isinstance(value.get("status"), str)
|
or value["stable_id"] != job["bvid"]
|
or value["creator_uid"] != job["creator_uid"]
|
or value["source_url"] != job["canonical_url"]
|
or value["published_at"] != job["published_at"]
|
or value["item_type"] != "video"
|
or _FORMAL_PRIOR_STATUS_RE.fullmatch(value["status"]) is None
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
return value["status"]
|
|
|
def _validate_legacy_integer_uid_formal_prior(
|
payload: bytes,
|
raw_lines: Sequence[bytes],
|
line_ordinal: int,
|
value: dict[str, Any],
|
config: HostConfig,
|
job: dict[str, Any],
|
) -> str:
|
_validate_formal_legacy_manifest()
|
if (
|
len(payload) < FORMAL_PREFIX_BYTES
|
or len(raw_lines) < FORMAL_PREFIX_LINES
|
or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper()
|
!= FORMAL_PREFIX_SHA256
|
or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
|
or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
expected = _FORMAL_LEGACY_INTEGER_UID_BY_LINE.get(line_ordinal)
|
if expected is None:
|
raise WorkerError("E_COMPLETION_FORMAL")
|
raw_line = raw_lines[line_ordinal - 1]
|
if (
|
len(raw_line) != expected[1]
|
or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
|
or not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
|
or type(value.get("schema_version")) is not int
|
or value["schema_version"] != 1
|
or value.get("creator") != config.creator_name
|
or type(value.get("creator_uid")) is not int
|
or value["creator_uid"] != expected[3]
|
or value.get("stable_id") != expected[4]
|
or value.get("source_url") != expected[5]
|
or value.get("published_at") != expected[6]
|
or value.get("item_type") != expected[7]
|
or value.get("status") != expected[8]
|
or str(expected[3]) != job["creator_uid"]
|
or expected[4] != job["bvid"]
|
or expected[5] != job["canonical_url"]
|
or expected[6] != job["published_at"]
|
):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
return expected[8]
|
|
|
def _read_jsonl_objects(path: Path, *, limit: int, error_code: str) -> tuple[bytes, list[dict[str, Any]]]:
|
try:
|
if path.exists():
|
parent = path.parent.resolve(strict=True)
|
_ordinary_exact_file(path, parent)
|
payload = path.read_bytes()
|
else:
|
payload = b""
|
if len(payload) > limit:
|
raise WorkerError(error_code)
|
complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
|
records: list[dict[str, Any]] = []
|
for raw_line in complete.splitlines():
|
if not raw_line:
|
raise WorkerError(error_code)
|
value = strict_json_loads(raw_line)
|
records.append(value)
|
return payload, records
|
except WorkerError:
|
raise
|
except (OSError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
|
raise WorkerError(error_code) from exc
|
|
|
def _append_jsonl_idempotent(
|
path: Path,
|
record: dict[str, Any],
|
*,
|
identity_key: str,
|
identity_value: str,
|
limit: int,
|
error_code: str,
|
) -> None:
|
expected = _canonical_json_line(record)
|
payload, records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
|
matches = [value for value in records if value.get(identity_key) == identity_value]
|
if len(matches) > 1 or (matches and matches[0] != record):
|
raise WorkerError("E_COMPLETION_REPLAY")
|
if matches:
|
return
|
suffix = b"" if not payload or payload.endswith(b"\n") else payload[payload.rfind(b"\n") + 1:]
|
if suffix and (len(suffix) >= len(expected) or expected[:len(suffix)] != suffix):
|
raise WorkerError(error_code)
|
parent = path.parent.resolve(strict=True)
|
try:
|
if not path.exists():
|
with path.open("xb") as created:
|
created.flush()
|
os.fsync(created.fileno())
|
_ordinary_exact_file(path, parent)
|
with path.open("r+b", buffering=0) as stream:
|
current = stream.read()
|
if current != payload:
|
raise WorkerError(error_code)
|
stream.seek(0, os.SEEK_END)
|
stream.write(expected[len(suffix):])
|
stream.flush()
|
os.fsync(stream.fileno())
|
_completion_test_seam(f"AFTER_{identity_key.upper()}_APPEND")
|
final_payload, final_records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
|
if not final_payload.endswith(b"\n") or sum(
|
value.get(identity_key) == identity_value and value == record for value in final_records
|
) != 1:
|
raise WorkerError(error_code)
|
except WorkerError:
|
raise
|
except OSError as exc:
|
raise WorkerError(error_code) from exc
|
|
|
@contextmanager
|
def _completion_lock(path: Path) -> Iterable[None]:
|
try:
|
path.parent.resolve(strict=True)
|
with path.open("a+b") as stream:
|
if stream.seek(0, os.SEEK_END) == 0:
|
stream.write(b"\0")
|
stream.flush()
|
os.fsync(stream.fileno())
|
_ordinary_exact_file(path, path.parent)
|
stream.seek(0)
|
if os.name == "nt":
|
import msvcrt # noqa: PLC0415
|
|
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
|
try:
|
yield
|
finally:
|
stream.seek(0)
|
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
|
else:
|
import fcntl # noqa: PLC0415
|
|
fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
try:
|
yield
|
finally:
|
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
|
except WorkerError:
|
raise
|
except OSError as exc:
|
raise WorkerError("E_COMPLETION_BUSY") from exc
|
|
|
def _commit_formal_and_handoff(
|
config: HostConfig,
|
job: dict[str, Any],
|
formal_name: str,
|
mapping_name: str,
|
persisted: dict[str, Any],
|
*,
|
media_complete_acknowledged: bool = False,
|
) -> None:
|
"""Idempotently close formal publication and processing handoff before COMPLETE."""
|
if (
|
not isinstance(config.queue_lock_path, Path)
|
or not isinstance(config.formal_manifest_path, Path)
|
or not isinstance(config.processing_handoff_path, Path)
|
or not isinstance(config.creator_name, str) or not config.creator_name
|
):
|
raise WorkerError("E_COMPLETION_CONFIG")
|
handoff_id = f"HANDOFF-BILI-MEDIA-{job['job_id'][:32].upper()}"
|
with _completion_lock(config.queue_lock_path):
|
formal_payload, formal_records = _read_jsonl_objects(
|
config.formal_manifest_path, limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL"
|
)
|
raw_lines = _formal_raw_lines(formal_payload)
|
if len(raw_lines) != len(formal_records):
|
raise WorkerError("E_COMPLETION_FORMAL")
|
prior_status = None
|
seen_prior_identities: set[tuple[str, str, str, str, str, str]] = set()
|
for line_ordinal, value in enumerate(formal_records, 1):
|
if value.get("queue_job_id") == job["job_id"]:
|
continue
|
stable_matches = value.get("stable_id") == job["bvid"]
|
source_matches = value.get("source_url") == job["canonical_url"]
|
if not stable_matches and not source_matches:
|
continue
|
item_type = value.get("item_type")
|
if item_type == "video_transcript" and stable_matches and source_matches:
|
continue
|
if not stable_matches or not source_matches:
|
raise WorkerError("E_COMPLETION_FORMAL")
|
missing = _FORMAL_PRIOR_REQUIRED_KEYS - set(value)
|
if line_ordinal in _FORMAL_LEGACY_INTEGER_UID_BY_LINE:
|
prior_status = _validate_legacy_integer_uid_formal_prior(
|
formal_payload, raw_lines, line_ordinal, value, config, job
|
)
|
creator_uid = job["creator_uid"]
|
elif missing == {"creator_uid"}:
|
prior_status = _validate_legacy_formal_prior(
|
formal_payload, raw_lines, line_ordinal, value, config, job
|
)
|
creator_uid = job["creator_uid"]
|
else:
|
prior_status = _validate_current_formal_prior(value, config, job)
|
creator_uid = value["creator_uid"]
|
prior_identity = (
|
value["stable_id"], creator_uid, value["source_url"],
|
value["published_at"], value["item_type"], prior_status,
|
)
|
if prior_identity in seen_prior_identities:
|
raise WorkerError("E_COMPLETION_FORMAL")
|
seen_prior_identities.add(prior_identity)
|
if prior_status is None and media_complete_acknowledged is not True:
|
# A first formal row has no historical status to supersede. It is
|
# permitted only after the typed Host ACK proves the governed queue
|
# job's exact MEDIA_COMPLETE identity is already durable.
|
raise WorkerError("E_COMPLETION_FORMAL")
|
formal_record = {
|
"schema_version": 1,
|
"creator": config.creator_name,
|
"creator_uid": job["creator_uid"],
|
"item_type": "video",
|
"stable_id": job["bvid"],
|
"title": job["title"],
|
"source_url": job["canonical_url"],
|
"published_at": job["published_at"],
|
"collected_at": persisted["completed_at"],
|
"status": "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT",
|
"video_path": str(config.destination / formal_name),
|
"mapping_path": str(config.destination / mapping_name),
|
"bytes": persisted["bytes"],
|
"sha256": persisted["sha256"],
|
"duration_seconds": persisted["duration_seconds"],
|
"video_codec": persisted["video_codec"],
|
"audio_codec": persisted["audio_codec"],
|
"processing_handoff_id": handoff_id,
|
"queue_job_id": job["job_id"],
|
}
|
if prior_status is not None:
|
formal_record["supersedes_status"] = prior_status
|
handoff_record = {
|
"schema": 1,
|
"type": "media-processing-handoff",
|
"status": "READY",
|
"handoff_id": handoff_id,
|
"queue_job_id": job["job_id"],
|
"creator_uid": job["creator_uid"],
|
"bvid": job["bvid"],
|
"source_url": job["canonical_url"],
|
"media_path": str(config.destination / formal_name),
|
"mapping_path": str(config.destination / mapping_name),
|
"bytes": persisted["bytes"],
|
"sha256": persisted["sha256"],
|
"duration_seconds": persisted["duration_seconds"],
|
"video_codec": persisted["video_codec"],
|
"audio_codec": persisted["audio_codec"],
|
"created_at": persisted["completed_at"],
|
}
|
_append_jsonl_idempotent(
|
config.formal_manifest_path, formal_record,
|
identity_key="queue_job_id", identity_value=job["job_id"],
|
limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL",
|
)
|
_completion_test_seam("BETWEEN_FORMAL_AND_HANDOFF")
|
_append_jsonl_idempotent(
|
config.processing_handoff_path, handoff_record,
|
identity_key="queue_job_id", identity_value=job["job_id"],
|
limit=8 * 1024 * 1024, error_code="E_COMPLETION_HANDOFF",
|
)
|
_completion_test_seam("BEFORE_COMPLETION_RETURN")
|
|
|
@dataclass
|
class _LockedPublishedFile:
|
"""Read-only handle whose sharing mode denies writers, deletion and replacement."""
|
|
path: Path
|
parent: Path
|
error_code: str
|
stream: Any
|
identity: tuple[int, ...]
|
|
@classmethod
|
def open(cls, path: Path, parent: Path, error_code: str) -> "_LockedPublishedFile":
|
stream: Any | None = None
|
try:
|
before = _ordinary_exact_file(path, parent)
|
if os.name != "nt":
|
# The deployed Host is Windows-only. Keep non-Windows imports
|
# fail-closed while retaining a no-follow advisory read lock.
|
import fcntl # noqa: PLC0415
|
|
flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
descriptor = os.open(path, flags)
|
try:
|
fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
|
stream = os.fdopen(descriptor, "rb", closefd=True)
|
except BaseException:
|
os.close(descriptor)
|
raise
|
else:
|
import msvcrt # noqa: PLC0415
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
create_file = kernel32.CreateFileW
|
create_file.argtypes = (
|
ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
)
|
create_file.restype = ctypes.c_void_p
|
close_handle = kernel32.CloseHandle
|
close_handle.argtypes = (ctypes.c_void_p,)
|
close_handle.restype = ctypes.c_int
|
handle = create_file(
|
str(path),
|
0x80000000, # GENERIC_READ
|
0x00000001, # FILE_SHARE_READ: deny write/delete/path replacement
|
None,
|
3, # OPEN_EXISTING
|
0x00000080 | 0x00200000 | 0x08000000,
|
# FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OPEN_REPARSE_POINT |
|
# FILE_FLAG_SEQUENTIAL_SCAN
|
None,
|
)
|
if handle in (None, ctypes.c_void_p(-1).value):
|
code = ctypes.get_last_error()
|
raise OSError(code, ctypes.FormatError(code), str(path))
|
descriptor: int | None = None
|
try:
|
descriptor = msvcrt.open_osfhandle(
|
int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0)
|
)
|
handle = None
|
stream = os.fdopen(descriptor, "rb", closefd=True)
|
descriptor = None
|
finally:
|
if descriptor is not None:
|
os.close(descriptor)
|
if handle is not None:
|
close_handle(handle)
|
after = _ordinary_exact_file(path, parent)
|
identity = _stable_file_identity(os.fstat(stream.fileno()))
|
if identity != _stable_file_identity(before) or identity != _stable_file_identity(after):
|
raise WorkerError(error_code)
|
return cls(path=path, parent=parent, error_code=error_code, stream=stream, identity=identity)
|
except BaseException as exc:
|
if stream is not None:
|
stream.close()
|
if isinstance(exc, WorkerError) and exc.code == error_code:
|
raise
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
raise
|
raise WorkerError(error_code) from exc
|
|
def close(self) -> None:
|
self.stream.close()
|
|
def assert_path_identity(self) -> os.stat_result:
|
try:
|
path_stat = _ordinary_exact_file(self.path, self.parent)
|
handle_stat = os.fstat(self.stream.fileno())
|
except (OSError, WorkerError) as exc:
|
raise WorkerError(self.error_code) from exc
|
if (
|
_stable_file_identity(path_stat) != self.identity
|
or _stable_file_identity(handle_stat) != self.identity
|
):
|
raise WorkerError(self.error_code)
|
return handle_stat
|
|
def read_all(self, maximum: int) -> bytes:
|
try:
|
self.stream.seek(0)
|
payload = self.stream.read(maximum + 1)
|
if len(payload) > maximum or self.stream.read(1) != b"":
|
raise WorkerError(self.error_code)
|
self.assert_path_identity()
|
return payload
|
except WorkerError:
|
raise
|
except OSError as exc:
|
raise WorkerError(self.error_code) from exc
|
|
def sha256(self) -> str:
|
digest = hashlib.sha256()
|
try:
|
self.stream.seek(0)
|
for chunk in iter(lambda: self.stream.read(1024 * 1024), b""):
|
digest.update(chunk)
|
self.assert_path_identity()
|
return digest.hexdigest()
|
except WorkerError:
|
raise
|
except OSError as exc:
|
raise WorkerError(self.error_code) from exc
|
|
|
def _bridge_number(value: Any) -> float:
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
number = float(value)
|
if not math.isfinite(number) or number < 0:
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
return number
|
|
|
def _bridge_remote_matches_expected(expected_duration_ms: int, remote: float) -> bool:
|
expected = expected_duration_ms / 1000
|
if abs(remote - expected) <= 0.001:
|
return True
|
return (
|
expected_duration_ms % 1000 == 0
|
and 0 < expected - remote < 1
|
and math.ceil(remote) == int(expected)
|
)
|
|
|
def _validate_complete_bridge_item(
|
item: Any,
|
job: dict[str, Any],
|
*,
|
persisted: bool,
|
) -> dict[str, Any]:
|
if not isinstance(item, dict):
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
expected_keys = _BRIDGE_MAPPING_KEYS if persisted else _BRIDGE_ITEM_KEYS
|
if set(item) != expected_keys:
|
if persisted or set(item) != _BRIDGE_ITEM_WARNING_KEYS:
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
warning = item["warning"]
|
if (
|
not isinstance(warning, str)
|
or _BRIDGE_CLEANUP_WARNING_RE.fullmatch(warning) is None
|
):
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
if not persisted and item["status"] != "COMPLETE":
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
if (
|
item["schema_version"] != "1.0"
|
or item["bvid"] != job["bvid"]
|
or item["source"] != job["canonical_url"]
|
or item["published_at"] != job["published_at"]
|
or item["local_file"] != f"{job['bvid']}.mkv"
|
or item["acquisition_mode"] != "authorized_browser_file_handoff"
|
or not isinstance(item["title"], str)
|
or not 1 <= len(item["title"]) <= 1024
|
or any(ord(character) < 0x20 for character in item["title"])
|
or not isinstance(item["bytes"], int)
|
or isinstance(item["bytes"], bool)
|
or item["bytes"] <= 0
|
or not isinstance(item["sha256"], str)
|
or _LOWER_SHA256_RE.fullmatch(item["sha256"]) is None
|
or item["handoff_source_sha256"] != item["sha256"]
|
or not isinstance(item["format_name"], str)
|
or "matroska" not in item["format_name"].split(",")
|
or not isinstance(item["video_codec"], str)
|
or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["video_codec"])
|
or not isinstance(item["audio_codec"], str)
|
or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["audio_codec"])
|
or not isinstance(item["completed_at"], str)
|
or _UTC_ISO_RE.fullmatch(item["completed_at"]) is None
|
):
|
raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
|
duration = _bridge_number(item["duration_seconds"])
|
remote = _bridge_number(item["remote_duration_seconds"])
|
local = _bridge_number(item["local_duration_seconds"])
|
delta = _bridge_number(item["duration_delta_seconds"])
|
tolerance = _bridge_number(item["duration_tolerance_seconds"])
|
expected_tolerance = max(3.0, remote * 0.001)
|
if (
|
not _bridge_remote_matches_expected(job["expected_duration_ms"], remote)
|
or abs(duration - local) > 1e-9
|
or abs(delta - abs(local - remote)) > 1e-9
|
or abs(tolerance - expected_tolerance) > 1e-9
|
or delta > tolerance
|
):
|
raise WorkerError("E_BRIDGE_DURATION_SHA")
|
return item
|
|
|
def _read_exact_published_bridge_result(
|
config: HostConfig,
|
job: dict[str, Any],
|
*,
|
expected_item: dict[str, Any] | None = None,
|
on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
|
on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
|
) -> tuple[str, str] | None:
|
destination = config.destination.resolve(strict=True)
|
formal_name = f"{job['bvid']}.mkv"
|
mapping_name = f"{job['bvid']}.download.json"
|
formal_path = destination / formal_name
|
mapping_path = destination / mapping_name
|
present = (formal_path.exists(), mapping_path.exists())
|
if present == (False, False):
|
return None
|
if present != (True, True):
|
raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
|
formal_lock = _LockedPublishedFile.open(formal_path, destination, "E_BRIDGE_DURATION_SHA")
|
try:
|
mapping_lock = _LockedPublishedFile.open(
|
mapping_path, destination, "E_BRIDGE_MAPPING_READBACK"
|
)
|
try:
|
_stable_pair_test_seam("LOCKS_ACQUIRED")
|
mapping_stat = mapping_lock.assert_path_identity()
|
if mapping_stat.st_size <= 0 or mapping_stat.st_size > 64 * 1024:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
mapping_bytes = mapping_lock.read_all(64 * 1024)
|
try:
|
persisted = _validate_complete_bridge_item(
|
strict_json_loads(mapping_bytes), job, persisted=True
|
)
|
except WorkerError as exc:
|
if exc.code == "E_BRIDGE_DURATION_SHA":
|
raise
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
|
except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
|
if expected_item is not None:
|
if {key: expected_item[key] for key in _BRIDGE_MAPPING_KEYS} != persisted:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
formal_stat = formal_lock.assert_path_identity()
|
formal_sha = formal_lock.sha256().casefold()
|
if persisted["bytes"] != formal_stat.st_size or persisted["sha256"] != formal_sha:
|
raise WorkerError("E_BRIDGE_DURATION_SHA")
|
_stable_pair_test_seam("AFTER_INITIAL_PAIR")
|
|
# The two Windows handles deny write/delete/path replacement while
|
# FFprobe opens its read-only view. Hashing and JSON parsing use these
|
# same handles, and the consumer commit executes before handle release.
|
probe_mkv(config.ffprobe, formal_path, job)
|
formal_lock.assert_path_identity()
|
mapping_lock.assert_path_identity()
|
final_formal_sha = formal_lock.sha256().casefold()
|
_stable_pair_test_seam("AFTER_FINAL_MEDIA_HASH")
|
final_mapping_bytes = mapping_lock.read_all(64 * 1024)
|
_stable_pair_test_seam("AFTER_FINAL_MAPPING_READ")
|
if final_formal_sha != persisted["sha256"]:
|
raise WorkerError("E_BRIDGE_DURATION_SHA")
|
if final_mapping_bytes != mapping_bytes:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
try:
|
final_persisted = _validate_complete_bridge_item(
|
strict_json_loads(final_mapping_bytes), job, persisted=True
|
)
|
except (WorkerError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
|
if final_persisted != persisted:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
formal_lock.assert_path_identity()
|
mapping_lock.assert_path_identity()
|
_stable_pair_test_seam("BEFORE_COMMIT")
|
media_identity = validate_media_complete_identity(
|
{
|
"formal_filename": formal_name,
|
"mapping_filename": mapping_name,
|
"media_bytes": persisted["bytes"],
|
"media_sha256": persisted["sha256"].upper(),
|
"mapping_bytes": len(final_mapping_bytes),
|
"mapping_sha256": hashlib.sha256(final_mapping_bytes).hexdigest().upper(),
|
"duration_milliseconds": int(round(
|
float(persisted["local_duration_seconds"]) * 1_000
|
)),
|
"video_codec": persisted["video_codec"],
|
"audio_codec": persisted["audio_codec"],
|
},
|
job,
|
)
|
if on_media_verified is not None:
|
on_media_verified(formal_name, mapping_name, persisted, media_identity)
|
if on_verified is not None:
|
on_verified(formal_name, mapping_name, persisted)
|
return formal_name, mapping_name
|
finally:
|
mapping_lock.close()
|
finally:
|
formal_lock.close()
|
|
|
def recover_published_task(
|
config: HostConfig,
|
job: dict[str, Any],
|
*,
|
cancel_check: Callable[[], bool],
|
report: Callable[..., None],
|
commit_begin: Callable[[], None],
|
) -> tuple[str, str]:
|
"""Verify and consume one exact published pair before secret transfer."""
|
if job["creator_uid"] not in config.creator_allowlist:
|
raise WorkerError("E_ALLOWLIST")
|
committed: tuple[str, str] | None = None
|
|
def media_verified(
|
_formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
|
media_identity: dict[str, Any],
|
) -> None:
|
report("MEDIA_COMPLETE", 100, media_identity)
|
|
def consume(formal_name: str, mapping_name: str, persisted: dict[str, Any]) -> None:
|
nonlocal committed
|
if cancel_check():
|
raise CancelRequested()
|
report("POSTPROCESS_PENDING", 100)
|
commit_begin()
|
_commit_formal_and_handoff(
|
config, job, formal_name, mapping_name, persisted,
|
media_complete_acknowledged=True,
|
)
|
committed = (formal_name, mapping_name)
|
|
recovered = _read_exact_published_bridge_result(
|
config, job, on_media_verified=media_verified, on_verified=consume
|
)
|
if recovered is None:
|
raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
|
if committed != recovered:
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
return recovered
|
|
|
def _bridge_command(
|
config: HostConfig, candidate: Path, job: dict[str, Any], batch_json: Path
|
) -> list[str]:
|
return [
|
str(config.bridge_python),
|
str(config.bridge_script),
|
"--input",
|
str(batch_json),
|
"accept-browser-file",
|
"--bvid",
|
job["bvid"],
|
"--media-file",
|
str(candidate),
|
"--destination",
|
str(config.destination),
|
"--ffprobe",
|
str(config.ffprobe),
|
"--expected-duration-ms",
|
str(job["expected_duration_ms"]),
|
]
|
|
|
def run_frozen_bridge(
|
config: HostConfig,
|
candidate: Path,
|
job: dict[str, Any],
|
*,
|
on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
|
on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
|
) -> tuple[str, str]:
|
started = time.monotonic()
|
|
def failure(code: str, state: str, reason: str) -> WorkerError:
|
return WorkerError(code, {
|
"attempts": 1,
|
"elapsed_ms": max(0, min(7_200_000, int((time.monotonic() - started) * 1000))),
|
"state": state,
|
"reason": reason,
|
})
|
|
batch_json = candidate.parent / "bridge-input.json"
|
batch_payload = {
|
"schema_version": "1.0",
|
"batch_id": f"generic-{job['job_id'][:16]}",
|
"items": [{
|
"bvid": job["bvid"],
|
"source_url": job["canonical_url"],
|
"published_at": job["published_at"],
|
"title": job["title"],
|
"expected_duration_ms": job["expected_duration_ms"],
|
}],
|
}
|
try:
|
with batch_json.open("xb") as stream:
|
stream.write((json.dumps(batch_payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8"))
|
stream.flush()
|
os.fsync(stream.fileno())
|
except BaseException as exc:
|
batch_json.unlink(missing_ok=True)
|
raise failure(
|
"E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "BATCH_RECEIPT_CREATE_FAILED"
|
) from exc
|
try:
|
result = _run_local(
|
_bridge_command(config, candidate, job, batch_json),
|
BRIDGE_TIMEOUT_SECONDS,
|
capture_stdout=True,
|
)
|
except subprocess.TimeoutExpired as exc:
|
raise failure(
|
"E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_TIMEOUT"
|
) from exc
|
except OSError as exc:
|
raise failure(
|
"E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_INVOCATION_FAILED"
|
) from exc
|
finally:
|
batch_json.unlink(missing_ok=True)
|
if result.returncode != 0:
|
try:
|
stopped = strict_json_loads(result.stdout)
|
except (ProtocolError, UnicodeError, json.JSONDecodeError):
|
stopped = None
|
bridge_stops = {
|
"E_BRIDGE_SOURCE_STABILITY": (
|
"BRIDGE_SOURCE_STABILITY", "SOURCE_FILE_INVALID"
|
),
|
"E_BRIDGE_METADATA_BINDING": (
|
"BRIDGE_METADATA_BINDING", "EXPECTED_METADATA_MISMATCH"
|
),
|
"E_BRIDGE_FFPROBE": (
|
"BRIDGE_FFPROBE", "LOCAL_MEDIA_PROBE_FAILED"
|
),
|
"E_BRIDGE_DURATION_SHA": (
|
"DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"
|
),
|
"E_BRIDGE_PUBLISH": (
|
"CREATE_NEW_PUBLISH", "BRIDGE_REPORTED_PUBLISH_FAILURE"
|
),
|
}
|
if (
|
isinstance(stopped, dict)
|
and set(stopped) == {"result", "error_code"}
|
and stopped.get("result") == "SAFETY_STOP"
|
and stopped.get("error_code") in bridge_stops
|
):
|
code = stopped["error_code"]
|
state, reason = bridge_stops[code]
|
raise failure(code, state, reason)
|
raise failure("E_BRIDGE_EXIT", "BRIDGE_EXIT", "NONZERO_EXIT")
|
|
def read_published(expected_item: dict[str, Any] | None) -> tuple[str, str]:
|
try:
|
published_value = _read_exact_published_bridge_result(
|
config, job, expected_item=expected_item,
|
on_media_verified=on_media_verified, on_verified=on_verified,
|
)
|
except WorkerError as exc:
|
if exc.code.startswith("E_COMPLETION_"):
|
raise
|
code = exc.code if exc.code.startswith("E_BRIDGE_") else "E_BRIDGE_MAPPING_READBACK"
|
state, reason = {
|
"E_BRIDGE_MEDIA_MAPPING": ("MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"),
|
"E_BRIDGE_DURATION_SHA": ("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"),
|
}.get(code, ("MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"))
|
raise failure(code, state, reason) from exc
|
if published_value is None:
|
raise failure(
|
"E_BRIDGE_MEDIA_MAPPING", "MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"
|
)
|
return published_value
|
|
def preserve_verified_media_before_output_failure(_cause: BaseException | None = None) -> bool:
|
"""Persist media truth, but do not commit postprocess through an invalid wire result."""
|
try:
|
if os.path.lexists(config.destination / ".bili-download-staging"):
|
raise failure(
|
"E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
|
)
|
published_value = _read_exact_published_bridge_result(
|
config, job, on_media_verified=on_media_verified
|
)
|
if published_value is None:
|
raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
|
return True
|
except WorkerError as exc:
|
if exc.code.startswith("E_COMPLETION_"):
|
raise
|
return False
|
|
try:
|
payload = strict_json_loads(result.stdout)
|
except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
|
preserve_verified_media_before_output_failure(exc)
|
raise failure(
|
"E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
|
) from exc
|
if (
|
not isinstance(payload, dict)
|
or set(payload) != {"batch_id", "command", "result", "success_count", "failure_count", "items"}
|
or payload.get("batch_id") != f"generic-{job['job_id'][:16]}"
|
or payload.get("result") != "PASS"
|
or payload.get("command") != "accept-browser-file"
|
or payload.get("success_count") != 1
|
or payload.get("failure_count") != 0
|
):
|
preserve_verified_media_before_output_failure()
|
raise failure(
|
"E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
|
)
|
items = payload.get("items")
|
if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict):
|
preserve_verified_media_before_output_failure()
|
raise failure(
|
"E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
|
)
|
try:
|
item = _validate_complete_bridge_item(items[0], job, persisted=False)
|
except WorkerError as exc:
|
preserve_verified_media_before_output_failure(exc)
|
code = exc.code if exc.code == "E_BRIDGE_DURATION_SHA" else "E_BRIDGE_OUTPUT_SCHEMA"
|
state, reason = (
|
("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH")
|
if code == "E_BRIDGE_DURATION_SHA"
|
else ("BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID")
|
)
|
raise failure(code, state, reason) from exc
|
if "warning" in item:
|
try:
|
if os.path.lexists(config.destination / ".bili-download-staging"):
|
raise failure(
|
"E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
|
)
|
except OSError as exc:
|
raise failure(
|
"E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
|
) from exc
|
return read_published(item)
|
|
|
def ytdlp_options(
|
config: HostConfig,
|
run_directory: Path,
|
cookie_stream: io.StringIO,
|
progress_hook: Callable[[dict[str, Any]], None],
|
) -> dict[str, Any]:
|
return {
|
"cookiefile": cookie_stream,
|
"format": "bestvideo+bestaudio/best",
|
"merge_output_format": "mkv",
|
"outtmpl": str(run_directory / "%(id)s.%(format_id)s.%(ext)s"),
|
"noplaylist": True,
|
"continuedl": False,
|
"overwrites": False,
|
"cachedir": False,
|
"quiet": True,
|
"no_warnings": True,
|
"logger": NullLogger(),
|
"progress_hooks": [progress_hook],
|
"postprocessor_hooks": [progress_hook],
|
"socket_timeout": SOCKET_TIMEOUT_SECONDS,
|
"extractor_retries": EXTRACTOR_RETRIES,
|
"retries": HTTP_RETRIES,
|
"fragment_retries": FRAGMENT_RETRIES,
|
"file_access_retries": FILE_ACCESS_RETRIES,
|
"retry_sleep_functions": {
|
"http": lambda _attempt: 1,
|
"fragment": lambda _attempt: 1,
|
"file_access": lambda _attempt: 1,
|
"extractor": lambda _attempt: 1,
|
},
|
"ffmpeg_location": str(config.ffmpeg),
|
"writethumbnail": False,
|
"writesubtitles": False,
|
"writeautomaticsub": False,
|
"writeinfojson": False,
|
"writedescription": False,
|
"writecomments": False,
|
"getcomments": False,
|
"allow_playlist_files": False,
|
"external_downloader": {},
|
}
|
|
|
def run_authenticated_task(
|
start: dict[str, Any],
|
config: HostConfig,
|
*,
|
cancel_check: Callable[[], bool],
|
report: Callable[..., None],
|
stage_root: Path | None = None,
|
prepared_run_directory: Path | None = None,
|
commit_begin: Callable[[], None] | None = None,
|
closure_report: Callable[[bool], None] | None = None,
|
recovery_required: bool = False,
|
) -> tuple[str, str, bool]:
|
"""Run the exact task. The caller must already own this worker in a job."""
|
validate_start(start)
|
job = start["job"]
|
if job["creator_uid"] not in config.creator_allowlist:
|
raise WorkerError("E_ALLOWLIST")
|
root = fixed_stage_root(job["bvid"]) if stage_root is None else stage_root.resolve()
|
if prepared_run_directory is None:
|
run_directory = prepare_run_directory(root)
|
else:
|
run_directory = prepared_run_directory.resolve(strict=True)
|
_ensure_within(run_directory, root)
|
if not run_directory.is_dir() or _is_reparse(run_directory):
|
raise WorkerError("E_STAGE")
|
cookie_stream: io.StringIO | None = None
|
cookie_closed = False
|
committed = False
|
outcome: tuple[str, str, bool] | None = None
|
try:
|
def checkpoint() -> None:
|
if cancel_check():
|
raise CancelRequested()
|
|
checkpoint()
|
def consume_recovered(
|
formal_name: str, mapping_name: str, persisted: dict[str, Any]
|
) -> None:
|
nonlocal cookie_closed, committed, outcome
|
checkpoint()
|
report("POSTPROCESS_PENDING", 100)
|
if commit_begin is not None:
|
commit_begin()
|
_commit_formal_and_handoff(
|
config, job, formal_name, mapping_name, persisted,
|
media_complete_acknowledged=True,
|
)
|
cookie_closed = True
|
committed = True
|
outcome = (formal_name, mapping_name, True)
|
|
def recovered_media_verified(
|
_formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
|
media_identity: dict[str, Any],
|
) -> None:
|
report("MEDIA_COMPLETE", 100, media_identity)
|
|
recovered = _read_exact_published_bridge_result(
|
config, job, on_media_verified=recovered_media_verified,
|
on_verified=consume_recovered,
|
)
|
if recovered is not None:
|
if outcome != (recovered[0], recovered[1], True):
|
raise WorkerError("E_BRIDGE_MAPPING_READBACK")
|
return outcome
|
if recovery_required:
|
raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
|
cookie_stream = build_cookie_stream(start)
|
yt_dlp, _ = bootstrap_ytdlp()
|
secret_values = [
|
value
|
for cookie in start["cookies"]
|
for value in (cookie["name"], cookie["value"])
|
]
|
policy = SubprocessPolicy(
|
run_directory,
|
{config.ffmpeg, config.ffprobe, config.bridge_python},
|
secret_values,
|
)
|
policy.install()
|
|
def progress_hook(status: dict[str, Any]) -> None:
|
if cancel_check():
|
raise CancelRequested()
|
downloaded = status.get("downloaded_bytes")
|
total = status.get("total_bytes") or status.get("total_bytes_estimate")
|
progress = 0
|
if isinstance(downloaded, (int, float)) and isinstance(total, (int, float)) and total > 0:
|
progress = max(0, min(95, int(float(downloaded) * 95 / float(total))))
|
report("DOWNLOADING", progress)
|
|
opts = ytdlp_options(config, run_directory, cookie_stream, progress_hook)
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
report("CHECKING", 0)
|
download_info, single, signed_urls = prepare_download_info(ydl, job)
|
policy.secrets.update(url.casefold() for url in signed_urls)
|
checkpoint()
|
ydl.process_info(download_info)
|
checkpoint()
|
report("MERGING", 96)
|
files = [item for item in run_directory.iterdir() if item.is_file()]
|
if single:
|
originals = [item for item in files if item.suffix.casefold() not in {".part", ".tmp"}]
|
if len(originals) != 1:
|
raise WorkerError("E_STAGE")
|
final = run_directory / "complete.mkv"
|
checkpoint()
|
remux_single_to_mkv(config.ffmpeg, originals[0], final)
|
checkpoint()
|
originals[0].unlink()
|
else:
|
mkv_candidates = [item for item in files if item.suffix.casefold() == ".mkv"]
|
if len(mkv_candidates) != 1:
|
raise WorkerError("E_STAGE")
|
final = run_directory / "complete.mkv"
|
if mkv_candidates[0] != final:
|
if final.exists():
|
raise WorkerError("E_COLLISION")
|
checkpoint()
|
mkv_candidates[0].rename(final)
|
checkpoint()
|
checkpoint()
|
candidate = validate_unique_candidate(run_directory)
|
checkpoint()
|
probe_mkv(config.ffprobe, candidate, job)
|
checkpoint()
|
cookie_closed = close_cookie_stream(cookie_stream)
|
cookie_stream = None
|
report("VALIDATING", 98)
|
checkpoint()
|
report("PUBLISHING", 99)
|
if commit_begin is not None:
|
commit_begin()
|
checkpoint()
|
completion_committed = False
|
|
def consume_published(
|
formal_name: str, mapping_name: str, persisted: dict[str, Any]
|
) -> None:
|
nonlocal completion_committed
|
checkpoint()
|
report("POSTPROCESS_PENDING", 100)
|
_commit_formal_and_handoff(
|
config, job, formal_name, mapping_name, persisted,
|
media_complete_acknowledged=True,
|
)
|
completion_committed = True
|
|
formal, mapping = run_frozen_bridge(
|
config, candidate, job,
|
on_media_verified=lambda _formal, _mapping, _persisted, media: report(
|
"MEDIA_COMPLETE", 100, media
|
),
|
on_verified=consume_published,
|
)
|
if not completion_committed:
|
raise WorkerError("E_COMPLETION_HANDOFF")
|
committed = True
|
outcome = (formal, mapping, cookie_closed)
|
finally:
|
active_error = sys.exc_info()[1]
|
if cookie_stream is not None:
|
cookie_closed = close_cookie_stream(cookie_stream)
|
if closure_report is not None:
|
closure_report(cookie_closed)
|
try:
|
cleanup_run_directory(run_directory, root)
|
except OSError as exc:
|
if not committed:
|
if active_error is not None:
|
active_error.add_note("stage cleanup failed")
|
else:
|
raise WorkerError("E_STAGE") from exc
|
if outcome is None:
|
raise WorkerError("E_WORKER")
|
return outcome
|
