"""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 hashlib
|
import importlib.util
|
import io
|
import json
|
import math
|
import os
|
import shutil
|
import subprocess
|
import sys
|
import time
|
import uuid
|
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,
|
CANONICAL_URL,
|
DURATION_TOLERANCE_MS,
|
EXPECTED_DURATION_MS,
|
EXTRACTOR_RETRIES,
|
FILE_ACCESS_RETRIES,
|
FRAGMENT_RETRIES,
|
HTTP_RETRIES,
|
SOCKET_TIMEOUT_SECONDS,
|
TARGET_BVID,
|
YTDLP_MODULE_SHA256,
|
YTDLP_VERSION,
|
)
|
from .protocol import ProtocolError, strict_json_loads, validate_start
|
|
FROZEN_BRIDGE_SHA256 = "749FC486B0F42315BD463F11771FE2A7C71CAB53DD9AC2E411CE82E1175DFF13"
|
|
|
class WorkerError(Exception):
|
def __init__(self, code: str) -> None:
|
super().__init__(code)
|
self.code = code
|
|
|
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:
|
ffmpeg: Path
|
ffprobe: Path
|
bridge_python: Path
|
bridge_script: Path
|
batch_json: Path
|
yt_dlp_executable: Path
|
destination: Path
|
|
@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",
|
"target",
|
"canonical_url",
|
"ffmpeg",
|
"ffmpeg_sha256",
|
"ffprobe",
|
"ffprobe_sha256",
|
"bridge_python",
|
"bridge_python_sha256",
|
"bridge_script",
|
"bridge_script_sha256",
|
"batch_json",
|
"batch_json_sha256",
|
"yt_dlp_executable",
|
"yt_dlp_executable_sha256",
|
"destination",
|
}
|
if not isinstance(raw, dict) or set(raw) != expected:
|
raise WorkerError("E_CONFIG")
|
if raw["schema"] != 1 or raw["target"] != TARGET_BVID or raw["canonical_url"] != CANONICAL_URL:
|
raise WorkerError("E_CONFIG")
|
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(
|
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,
|
batch_json=cls._safe_absolute_file(raw["batch_json"], raw["batch_json_sha256"]),
|
yt_dlp_executable=cls._safe_absolute_file(
|
raw["yt_dlp_executable"], raw["yt_dlp_executable_sha256"]
|
),
|
destination=destination,
|
)
|
|
|
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() -> Path:
|
local_app_data = validated_local_app_data()
|
logical_root = (
|
local_app_data
|
/ "project-info"
|
/ "bili-auth-ingress"
|
/ TARGET_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 = fixed_stage_root() 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 | None = None) -> Path:
|
stage_root = fixed_stage_root() if root is None else 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 | None = None) -> Path:
|
"""Clean stale runs and create the secret-free task lease."""
|
stage_root = fixed_stage_root() if root is None else root.resolve()
|
cleanup_stale_runs(stage_root, boundary=stage_root if root is not None else None)
|
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) -> dict[str, Any]:
|
if not isinstance(info, dict) or info.get("id") != TARGET_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")
|
duration_ms = round(_finite_number(info.get("duration")) * 1000)
|
if abs(duration_ms - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_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],
|
*,
|
downloader_resolver: Callable[..., Any] | None = None,
|
) -> tuple[list[dict[str, Any]], bool]:
|
if not isinstance(download_info, dict) or download_info.get("id") != TARGET_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,
|
*,
|
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(CANONICAL_URL, download=False, process=True)
|
validate_processed_info(processed)
|
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,
|
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."""
|
|
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}
|
|
def check(self, event: str, arguments: tuple[Any, ...]) -> None:
|
if event != "subprocess.Popen":
|
return
|
executable, argv, _cwd, environment = arguments
|
if not isinstance(executable, (str, os.PathLike)) or not isinstance(argv, (list, tuple)):
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
executable_key = os.path.normcase(str(Path(executable).resolve()))
|
if executable_key not in self.executables:
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
text_args = [str(item) for item in argv]
|
folded = "\x00".join(text_args).casefold()
|
forbidden = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
|
if any(item in folded for item in forbidden) or any(item in folded for item in self.secrets):
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
if environment is not None:
|
encoded_env = "\x00".join(f"{key}={value}" for key, value in environment.items()).casefold()
|
if any(item in encoded_env for item in self.secrets):
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
if executable_key.endswith("ffmpeg.exe") or executable_key.endswith("ffprobe.exe"):
|
for index, argument in enumerate(text_args[:-1]):
|
if argument == "-i":
|
self._local_path(text_args[index + 1], must_exist=True)
|
output = text_args[-1]
|
if output not in {"-", "NUL"} and not output.startswith("-"):
|
self._local_path(output, must_exist=False)
|
|
def _local_path(self, value: str, *, must_exist: bool) -> Path:
|
path = Path(value)
|
if not path.is_absolute() or str(path).startswith("\\\\"):
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
try:
|
resolved = path.resolve(strict=must_exist)
|
except OSError as exc:
|
raise WorkerError("E_SUBPROCESS_POLICY") from exc
|
try:
|
resolved.relative_to(self.run_root)
|
except ValueError as exc:
|
raise WorkerError("E_SUBPROCESS_POLICY") from exc
|
if must_exist and (not resolved.is_file() or resolved.is_symlink()):
|
raise WorkerError("E_SUBPROCESS_POLICY")
|
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) -> None:
|
try:
|
result = _run_local(
|
[
|
str(ffprobe),
|
"-v",
|
"error",
|
"-show_entries",
|
"format=format_name: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")
|
|
|
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 _bridge_command(config: HostConfig, candidate: Path) -> list[str]:
|
return [
|
str(config.bridge_python),
|
str(config.bridge_script),
|
"--input",
|
str(config.batch_json),
|
"--yt-dlp",
|
str(config.yt_dlp_executable),
|
"accept-browser-file",
|
"--bvid",
|
TARGET_BVID,
|
"--media-file",
|
str(candidate),
|
"--destination",
|
str(config.destination),
|
"--ffprobe",
|
str(config.ffprobe),
|
]
|
|
|
def run_frozen_bridge(config: HostConfig, candidate: Path) -> tuple[str, str]:
|
try:
|
result = _run_local(
|
_bridge_command(config, candidate),
|
BRIDGE_TIMEOUT_SECONDS,
|
capture_stdout=True,
|
)
|
except subprocess.TimeoutExpired as exc:
|
raise WorkerError("E_BACKHALF") from exc
|
if result.returncode != 0:
|
raise WorkerError("E_BACKHALF")
|
try:
|
payload = json.loads(result.stdout.decode("utf-8"))
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
raise WorkerError("E_BACKHALF") from exc
|
if not isinstance(payload, dict) or payload.get("result") != "PASS":
|
raise WorkerError("E_BACKHALF")
|
items = payload.get("items")
|
if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict):
|
raise WorkerError("E_BACKHALF")
|
item = items[0]
|
required = {"bvid", "local_file", "bytes", "sha256", "duration_seconds", "acquisition_mode"}
|
if not required.issubset(item) or item["bvid"] != TARGET_BVID:
|
raise WorkerError("E_BACKHALF")
|
if item["acquisition_mode"] != "authorized_browser_file_handoff":
|
raise WorkerError("E_BACKHALF")
|
formal = item["local_file"]
|
mapping = f"{TARGET_BVID}.download.json"
|
if formal != f"{TARGET_BVID}.mkv":
|
raise WorkerError("E_BACKHALF")
|
formal_path = config.destination / formal
|
mapping_path = config.destination / mapping
|
if not formal_path.is_file() or not mapping_path.is_file():
|
raise WorkerError("E_BACKHALF")
|
try:
|
duration_ms = round(_finite_number(item["duration_seconds"]) * 1000)
|
except WorkerError as exc:
|
raise WorkerError("E_BACKHALF") from exc
|
if abs(duration_ms - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
|
raise WorkerError("E_BACKHALF")
|
formal_sha = sha256_file(formal_path)
|
if (
|
isinstance(item["bytes"], bool)
|
or not isinstance(item["bytes"], int)
|
or item["bytes"] != formal_path.stat().st_size
|
or not isinstance(item["sha256"], str)
|
or item["sha256"] != formal_sha
|
):
|
raise WorkerError("E_BACKHALF")
|
try:
|
persisted = strict_json_loads(mapping_path.read_bytes())
|
except (OSError, ProtocolError) as exc:
|
raise WorkerError("E_BACKHALF") from exc
|
matched_fields = {
|
"bvid": TARGET_BVID,
|
"source": CANONICAL_URL,
|
"local_file": formal,
|
"bytes": item["bytes"],
|
"sha256": formal_sha,
|
"acquisition_mode": "authorized_browser_file_handoff",
|
}
|
if any(persisted.get(key) != expected for key, expected in matched_fields.items()):
|
raise WorkerError("E_BACKHALF")
|
try:
|
persisted_duration_ms = round(_finite_number(persisted.get("duration_seconds")) * 1000)
|
except WorkerError as exc:
|
raise WorkerError("E_BACKHALF") from exc
|
if persisted_duration_ms != duration_ms:
|
raise WorkerError("E_BACKHALF")
|
return formal, mapping
|
|
|
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[[str, int], 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,
|
) -> tuple[str, str, bool]:
|
"""Run the exact task. The caller must already own this worker in a job."""
|
validate_start(start)
|
root = fixed_stage_root() 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()
|
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)
|
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)
|
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()
|
formal, mapping = run_frozen_bridge(config, candidate)
|
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
|
