"""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