From 917671c2207dd9cd22f229a594c42d7a71074361 Mon Sep 17 00:00:00 2001 From: Jesus <2573236+jalvarado-it@users.noreply.github.com> Date: Mon, 30 Mar 2026 18:43:12 -0600 Subject: [PATCH] Refactor app bootstrap and harden container configuration Extract environment-backed settings into a dedicated module and switch the aiohttp app to a factory-based bootstrap. Move runtime dependencies into typed app state, tighten CORS defaults, reject sensitive inline yt-dlp options, and harden the container entrypoint validation and ownership flow. Update tests to cover settings validation, app factory behavior, CORS policy, and entrypoint safety checks. --- .gitignore | 3 + app/config.py | 407 ++++++++++++++ app/main.py | 998 ++++++++++++++++++--------------- app/tests/conftest.py | 2 +- app/tests/test_api.py | 199 +++++-- app/tests/test_config.py | 87 ++- app/tests/test_entrypoint.py | 145 +++++ app/tests/test_main_helpers.py | 7 +- docker-entrypoint.sh | 89 ++- 9 files changed, 1372 insertions(+), 565 deletions(-) create mode 100644 app/config.py create mode 100644 app/tests/test_entrypoint.py diff --git a/.gitignore b/.gitignore index 0c58568..db1b798 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,6 @@ pending* __pycache__ .venv + +# Testing +./local \ No newline at end of file diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..58dbb08 --- /dev/null +++ b/app/config.py @@ -0,0 +1,407 @@ +from __future__ import annotations + +import json +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, ClassVar, Mapping +from urllib.parse import urlparse + +log = logging.getLogger("config") + + +class SettingsError(ValueError): + """Raised when environment-backed configuration is invalid.""" + + +@dataclass(frozen=True) +class Settings: + DOWNLOAD_DIR: str + AUDIO_DOWNLOAD_DIR: str + TEMP_DIR: str + DOWNLOAD_DIRS_INDEXABLE: bool + CUSTOM_DIRS: bool + CREATE_CUSTOM_DIRS: bool + CUSTOM_DIRS_EXCLUDE_REGEX: str + DELETE_FILE_ON_TRASHCAN: bool + STATE_DIR: str + URL_PREFIX: str + PUBLIC_HOST_URL: str + PUBLIC_HOST_AUDIO_URL: str + OUTPUT_TEMPLATE: str + OUTPUT_TEMPLATE_CHAPTER: str + OUTPUT_TEMPLATE_PLAYLIST: str + OUTPUT_TEMPLATE_CHANNEL: str + DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT: int + CLEAR_COMPLETED_AFTER: int + YTDL_OPTIONS: dict[str, Any] = field(repr=False) + YTDL_OPTIONS_FILE: str + ROBOTS_TXT: str + HOST: str + PORT: int + HTTPS: bool + CERTFILE: str + KEYFILE: str + DEFAULT_THEME: str + MAX_CONCURRENT_DOWNLOADS: int + LOGLEVEL: str + ENABLE_ACCESSLOG: bool + TRUSTED_ORIGINS: tuple[str, ...] + APP_ROOT: Path = field(repr=False, compare=False) + UI_DIST_DIR: Path = field(repr=False, compare=False) + ROBOTS_TXT_PATH: Path | None = field(repr=False, compare=False) + COOKIES_PATH: Path = field(repr=False, compare=False) + _inline_ytdl_options: dict[str, Any] = field(repr=False, compare=False) + _runtime_overrides: dict[str, Any] = field(default_factory=dict, repr=False, compare=False) + + _DEFAULTS: ClassVar[dict[str, str]] = { + "DOWNLOAD_DIR": ".", + "AUDIO_DOWNLOAD_DIR": "%%DOWNLOAD_DIR", + "TEMP_DIR": "%%DOWNLOAD_DIR", + "DOWNLOAD_DIRS_INDEXABLE": "false", + "CUSTOM_DIRS": "true", + "CREATE_CUSTOM_DIRS": "true", + "CUSTOM_DIRS_EXCLUDE_REGEX": r"(^|/)[.@].*$", + "DELETE_FILE_ON_TRASHCAN": "false", + "STATE_DIR": ".", + "URL_PREFIX": "", + "PUBLIC_HOST_URL": "download/", + "PUBLIC_HOST_AUDIO_URL": "audio_download/", + "OUTPUT_TEMPLATE": "%(title)s.%(ext)s", + "OUTPUT_TEMPLATE_CHAPTER": "%(title)s - %(section_number)02d - %(section_title)s.%(ext)s", + "OUTPUT_TEMPLATE_PLAYLIST": "%(playlist_title)s/%(title)s.%(ext)s", + "OUTPUT_TEMPLATE_CHANNEL": "%(channel)s/%(title)s.%(ext)s", + "DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT": "0", + "CLEAR_COMPLETED_AFTER": "0", + "YTDL_OPTIONS": "{}", + "YTDL_OPTIONS_FILE": "", + "ROBOTS_TXT": "", + "HOST": "0.0.0.0", + "PORT": "8081", + "HTTPS": "false", + "CERTFILE": "", + "KEYFILE": "", + "DEFAULT_THEME": "auto", + "MAX_CONCURRENT_DOWNLOADS": "3", + "LOGLEVEL": "INFO", + "ENABLE_ACCESSLOG": "false", + "TRUSTED_ORIGINS": "", + } + _BOOLEAN: ClassVar[tuple[str, ...]] = ( + "DOWNLOAD_DIRS_INDEXABLE", + "CUSTOM_DIRS", + "CREATE_CUSTOM_DIRS", + "DELETE_FILE_ON_TRASHCAN", + "HTTPS", + "ENABLE_ACCESSLOG", + ) + _INTEGER: ClassVar[tuple[str, ...]] = ( + "PORT", + "DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT", + "CLEAR_COMPLETED_AFTER", + "MAX_CONCURRENT_DOWNLOADS", + ) + _FRONTEND_KEYS: ClassVar[tuple[str, ...]] = ( + "CUSTOM_DIRS", + "CREATE_CUSTOM_DIRS", + "OUTPUT_TEMPLATE_CHAPTER", + "PUBLIC_HOST_URL", + "PUBLIC_HOST_AUDIO_URL", + "DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT", + ) + _SENSITIVE_INLINE_YTDL_KEYS: ClassVar[frozenset[str]] = frozenset( + { + "ap_mso", + "ap_password", + "ap_username", + "client_certificate", + "client_certificate_key", + "client_certificate_password", + "cookiefile", + "cookiesfrombrowser", + "netrc", + "netrc_cmd", + "netrc_location", + "password", + "username", + } + ) + _VALID_THEMES: ClassVar[frozenset[str]] = frozenset({"auto", "dark", "light"}) + + @classmethod + def from_env( + cls, + env: Mapping[str, str] | None = None, + *, + app_root: Path | None = None, + ) -> "Settings": + source = os.environ if env is None else env + raw = cls._resolve_defaults(source) + resolved_app_root = (app_root or Path(__file__).resolve().parent.parent).resolve() + ui_dist_dir = resolved_app_root / "ui" / "dist" / "metube" / "browser" + + raw["URL_PREFIX"] = cls._normalize_url_prefix(raw["URL_PREFIX"]) + inline_ytdl_options = cls._parse_ytdl_options(raw["YTDL_OPTIONS"]) + cls._validate_inline_ytdl_options(inline_ytdl_options) + + download_dir = cls._resolve_directory(raw["DOWNLOAD_DIR"], "DOWNLOAD_DIR") + audio_download_dir = cls._resolve_directory(raw["AUDIO_DOWNLOAD_DIR"], "AUDIO_DOWNLOAD_DIR") + temp_dir = cls._resolve_directory(raw["TEMP_DIR"], "TEMP_DIR") + state_dir = cls._resolve_directory(raw["STATE_DIR"], "STATE_DIR") + + ytdl_options_file = cls._resolve_optional_file(raw["YTDL_OPTIONS_FILE"], "YTDL_OPTIONS_FILE") + certfile = cls._resolve_optional_file(raw["CERTFILE"], "CERTFILE") + keyfile = cls._resolve_optional_file(raw["KEYFILE"], "KEYFILE") + robots_txt_path = cls._resolve_optional_file( + raw["ROBOTS_TXT"], + "ROBOTS_TXT", + base_dir=resolved_app_root, + ) + + parsed = { + "DOWNLOAD_DIR": str(download_dir), + "AUDIO_DOWNLOAD_DIR": str(audio_download_dir), + "TEMP_DIR": str(temp_dir), + "DOWNLOAD_DIRS_INDEXABLE": cls._parse_bool("DOWNLOAD_DIRS_INDEXABLE", raw["DOWNLOAD_DIRS_INDEXABLE"]), + "CUSTOM_DIRS": cls._parse_bool("CUSTOM_DIRS", raw["CUSTOM_DIRS"]), + "CREATE_CUSTOM_DIRS": cls._parse_bool("CREATE_CUSTOM_DIRS", raw["CREATE_CUSTOM_DIRS"]), + "CUSTOM_DIRS_EXCLUDE_REGEX": raw["CUSTOM_DIRS_EXCLUDE_REGEX"], + "DELETE_FILE_ON_TRASHCAN": cls._parse_bool("DELETE_FILE_ON_TRASHCAN", raw["DELETE_FILE_ON_TRASHCAN"]), + "STATE_DIR": str(state_dir), + "URL_PREFIX": raw["URL_PREFIX"], + "PUBLIC_HOST_URL": raw["PUBLIC_HOST_URL"], + "PUBLIC_HOST_AUDIO_URL": raw["PUBLIC_HOST_AUDIO_URL"], + "OUTPUT_TEMPLATE": raw["OUTPUT_TEMPLATE"], + "OUTPUT_TEMPLATE_CHAPTER": raw["OUTPUT_TEMPLATE_CHAPTER"], + "OUTPUT_TEMPLATE_PLAYLIST": raw["OUTPUT_TEMPLATE_PLAYLIST"], + "OUTPUT_TEMPLATE_CHANNEL": raw["OUTPUT_TEMPLATE_CHANNEL"], + "DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT": cls._parse_int( + "DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT", + raw["DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT"], + minimum=0, + ), + "CLEAR_COMPLETED_AFTER": cls._parse_int( + "CLEAR_COMPLETED_AFTER", + raw["CLEAR_COMPLETED_AFTER"], + minimum=0, + ), + "YTDL_OPTIONS": {}, + "YTDL_OPTIONS_FILE": str(ytdl_options_file) if ytdl_options_file else "", + "ROBOTS_TXT": str(robots_txt_path) if robots_txt_path else "", + "HOST": cls._parse_non_empty("HOST", raw["HOST"]), + "PORT": cls._parse_int("PORT", raw["PORT"], minimum=1), + "HTTPS": cls._parse_bool("HTTPS", raw["HTTPS"]), + "CERTFILE": str(certfile) if certfile else "", + "KEYFILE": str(keyfile) if keyfile else "", + "DEFAULT_THEME": cls._parse_theme(raw["DEFAULT_THEME"]), + "MAX_CONCURRENT_DOWNLOADS": cls._parse_int( + "MAX_CONCURRENT_DOWNLOADS", + raw["MAX_CONCURRENT_DOWNLOADS"], + minimum=1, + ), + "LOGLEVEL": cls._parse_loglevel(raw["LOGLEVEL"]), + "ENABLE_ACCESSLOG": cls._parse_bool("ENABLE_ACCESSLOG", raw["ENABLE_ACCESSLOG"]), + "TRUSTED_ORIGINS": cls._parse_trusted_origins(raw["TRUSTED_ORIGINS"]), + "APP_ROOT": resolved_app_root, + "UI_DIST_DIR": ui_dist_dir, + "ROBOTS_TXT_PATH": robots_txt_path, + "COOKIES_PATH": state_dir / "cookies.txt", + "_inline_ytdl_options": inline_ytdl_options, + } + + cls._validate_https(parsed["HTTPS"], certfile, keyfile) + cls._validate_ui_dist(ui_dist_dir) + + settings = cls(**parsed) + success, message = settings.load_ytdl_options() + if not success: + raise SettingsError(message) + return settings + + def frontend_safe(self) -> dict[str, Any]: + return {key: getattr(self, key) for key in self._FRONTEND_KEYS} + + def set_runtime_override(self, key: str, value: Any) -> None: + self._runtime_overrides[key] = value + self.YTDL_OPTIONS[key] = value + + def remove_runtime_override(self, key: str) -> None: + self._runtime_overrides.pop(key, None) + self.YTDL_OPTIONS.pop(key, None) + + def load_ytdl_options(self) -> tuple[bool, str]: + options = dict(self._inline_ytdl_options) + if self.YTDL_OPTIONS_FILE: + path = Path(self.YTDL_OPTIONS_FILE) + log.info('Loading yt-dlp custom options from "%s"', path) + if not path.exists(): + return (False, f'File "{path}" not found') + try: + with path.open(encoding="utf-8") as json_data: + file_options = json.load(json_data) + if not isinstance(file_options, dict): + raise TypeError("YTDL_OPTIONS_FILE must contain a JSON object") + except (OSError, TypeError, json.JSONDecodeError): + return (False, "YTDL_OPTIONS_FILE contents is invalid") + options.update(file_options) + + options.update(self._runtime_overrides) + self.YTDL_OPTIONS.clear() + self.YTDL_OPTIONS.update(options) + return (True, "") + + @classmethod + def _resolve_defaults(cls, env: Mapping[str, str]) -> dict[str, str]: + values = {key: str(env.get(key, default)) for key, default in cls._DEFAULTS.items()} + for key, value in list(values.items()): + if value.startswith("%%"): + values[key] = values[value[2:]] + return values + + @staticmethod + def _parse_bool(name: str, value: str) -> bool: + if value not in ("true", "false", "True", "False", "on", "off", "1", "0"): + raise SettingsError(f'Environment variable "{name}" is set to a non-boolean value "{value}"') + return value in ("true", "True", "on", "1") + + @staticmethod + def _parse_int(name: str, value: str, *, minimum: int | None = None) -> int: + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise SettingsError(f'Environment variable "{name}" must be an integer') from exc + if minimum is not None and parsed < minimum: + raise SettingsError(f'Environment variable "{name}" must be >= {minimum}') + return parsed + + @staticmethod + def _parse_non_empty(name: str, value: str) -> str: + stripped = str(value).strip() + if not stripped: + raise SettingsError(f'Environment variable "{name}" must not be empty') + return stripped + + @classmethod + def _parse_loglevel(cls, value: str) -> str: + parsed = getattr(logging, str(value).upper(), None) + if not isinstance(parsed, int): + raise SettingsError(f'Environment variable "LOGLEVEL" is invalid: "{value}"') + return str(value).upper() + + @classmethod + def _parse_theme(cls, value: str) -> str: + theme = str(value).strip().lower() + if theme not in cls._VALID_THEMES: + raise SettingsError('Environment variable "DEFAULT_THEME" must be one of auto, dark, light') + return theme + + @staticmethod + def _normalize_url_prefix(value: str) -> str: + prefix = str(value or "").strip() + if not prefix: + return "/" + if not prefix.startswith("/"): + prefix = f"/{prefix}" + if not prefix.endswith("/"): + prefix = f"{prefix}/" + return prefix + + @staticmethod + def _resolve_path(value: str, *, base_dir: Path | None = None) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + anchor = base_dir or Path.cwd() + path = anchor / path + return path.resolve(strict=False) + + @classmethod + def _resolve_directory(cls, value: str, name: str) -> Path: + path = cls._resolve_path(value) + if not path.exists(): + raise SettingsError(f'Configured path for "{name}" does not exist: {path}') + if not path.is_dir(): + raise SettingsError(f'Configured path for "{name}" is not a directory: {path}') + if not os.access(path, os.R_OK | os.W_OK | os.X_OK): + raise SettingsError(f'Configured path for "{name}" is not readable/writable: {path}') + return path + + @classmethod + def _resolve_optional_file( + cls, + value: str, + name: str, + *, + base_dir: Path | None = None, + ) -> Path | None: + if not str(value).strip(): + return None + path = cls._resolve_path(value, base_dir=base_dir) + if not path.exists(): + raise SettingsError(f'Configured path for "{name}" does not exist: {path}') + if not path.is_file(): + raise SettingsError(f'Configured path for "{name}" is not a file: {path}') + if not os.access(path, os.R_OK): + raise SettingsError(f'Configured path for "{name}" is not readable: {path}') + return path + + @staticmethod + def _parse_ytdl_options(value: str) -> dict[str, Any]: + try: + options = json.loads(value or "{}") + except json.JSONDecodeError as exc: + raise SettingsError("Environment variable YTDL_OPTIONS is invalid") from exc + if not isinstance(options, dict): + raise SettingsError("Environment variable YTDL_OPTIONS is invalid") + return options + + @classmethod + def _validate_inline_ytdl_options(cls, options: Mapping[str, Any]) -> None: + sensitive = sorted(cls._SENSITIVE_INLINE_YTDL_KEYS.intersection(options.keys())) + if sensitive: + raise SettingsError( + "Sensitive yt-dlp options are not allowed in YTDL_OPTIONS; " + f"use YTDL_OPTIONS_FILE instead ({', '.join(sensitive)})" + ) + + @staticmethod + def _validate_https(enabled: bool, certfile: Path | None, keyfile: Path | None) -> None: + if enabled and (certfile is None or keyfile is None): + raise SettingsError('HTTPS requires both "CERTFILE" and "KEYFILE"') + + @staticmethod + def _validate_ui_dist(ui_dist_dir: Path) -> None: + index_file = ui_dist_dir / "index.html" + if not index_file.exists(): + raise SettingsError( + "Could not find the frontend UI static assets. " + "Please run `node_modules/.bin/ng build` inside the ui folder" + ) + + @classmethod + def _parse_trusted_origins(cls, value: str) -> tuple[str, ...]: + raw_items: list[str] + stripped = str(value).strip() + if not stripped: + return () + if stripped.startswith("["): + try: + parsed = json.loads(stripped) + except json.JSONDecodeError as exc: + raise SettingsError('Environment variable "TRUSTED_ORIGINS" is invalid') from exc + if not isinstance(parsed, list): + raise SettingsError('Environment variable "TRUSTED_ORIGINS" is invalid') + raw_items = [str(item).strip() for item in parsed if str(item).strip()] + else: + raw_items = [item.strip() for item in stripped.split(",") if item.strip()] + + origins: list[str] = [] + for item in raw_items: + parsed = urlparse(item) + if not parsed.scheme or not parsed.netloc: + raise SettingsError(f'Environment variable "TRUSTED_ORIGINS" contains an invalid origin: "{item}"') + if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment: + raise SettingsError(f'Environment variable "TRUSTED_ORIGINS" must contain origins only: "{item}"') + normalized = f"{parsed.scheme}://{parsed.netloc}" + origins.append(normalized) + return tuple(dict.fromkeys(origins)) diff --git a/app/main.py b/app/main.py index db7be18..da92543 100644 --- a/app/main.py +++ b/app/main.py @@ -1,212 +1,104 @@ #!/usr/bin/env python3 # pylint: disable=no-member,method-hidden -import os -import sys +from __future__ import annotations + import asyncio -from pathlib import Path -from aiohttp import web -from aiohttp.log import access_logger -import ssl -import socket -import socketio -import logging import json +import logging +import os import pathlib import re -from watchfiles import DefaultFilter, Change, awatch +import socket +import ssl +import sys +from contextlib import suppress +from typing import Any -from ytdl import DownloadQueueNotifier, DownloadQueue, Download +import socketio +from aiohttp import web +from aiohttp.log import access_logger +from watchfiles import Change, DefaultFilter, awatch +from yarl import URL + +from config import Settings, SettingsError +from ytdl import Download, DownloadQueue, DownloadQueueNotifier from yt_dlp.version import __version__ as yt_dlp_version -log = logging.getLogger('main') +log = logging.getLogger("main") + def parseLogLevel(logLevel): if not isinstance(logLevel, str): return None return getattr(logging, logLevel.upper(), None) -# Configure logging before Config() uses it so early messages are not dropped. -# Only configure if no handlers are set (avoid clobbering hosting app settings). -if not logging.getLogger().hasHandlers(): - logging.basicConfig(level=parseLogLevel(os.environ.get('LOGLEVEL', 'INFO')) or logging.INFO) -class Config: - _DEFAULTS = { - 'DOWNLOAD_DIR': '.', - 'AUDIO_DOWNLOAD_DIR': '%%DOWNLOAD_DIR', - 'TEMP_DIR': '%%DOWNLOAD_DIR', - 'DOWNLOAD_DIRS_INDEXABLE': 'false', - 'CUSTOM_DIRS': 'true', - 'CREATE_CUSTOM_DIRS': 'true', - 'CUSTOM_DIRS_EXCLUDE_REGEX': r'(^|/)[.@].*$', - 'DELETE_FILE_ON_TRASHCAN': 'false', - 'STATE_DIR': '.', - 'URL_PREFIX': '', - 'PUBLIC_HOST_URL': 'download/', - 'PUBLIC_HOST_AUDIO_URL': 'audio_download/', - 'OUTPUT_TEMPLATE': '%(title)s.%(ext)s', - 'OUTPUT_TEMPLATE_CHAPTER': '%(title)s - %(section_number)02d - %(section_title)s.%(ext)s', - 'OUTPUT_TEMPLATE_PLAYLIST': '%(playlist_title)s/%(title)s.%(ext)s', - 'OUTPUT_TEMPLATE_CHANNEL': '%(channel)s/%(title)s.%(ext)s', - 'DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT' : '0', - 'CLEAR_COMPLETED_AFTER': '0', - 'YTDL_OPTIONS': '{}', - 'YTDL_OPTIONS_FILE': '', - 'ROBOTS_TXT': '', - 'HOST': '0.0.0.0', - 'PORT': '8081', - 'HTTPS': 'false', - 'CERTFILE': '', - 'KEYFILE': '', - 'BASE_DIR': '', - 'DEFAULT_THEME': 'auto', - 'MAX_CONCURRENT_DOWNLOADS': '3', - 'LOGLEVEL': 'INFO', - 'ENABLE_ACCESSLOG': 'false', - } +def configure_logging(raw_loglevel: str | None = None) -> None: + level = parseLogLevel(raw_loglevel or "INFO") or logging.INFO + if not logging.getLogger().hasHandlers(): + logging.basicConfig(level=level) + logging.getLogger().setLevel(level) - _BOOLEAN = ('DOWNLOAD_DIRS_INDEXABLE', 'CUSTOM_DIRS', 'CREATE_CUSTOM_DIRS', 'DELETE_FILE_ON_TRASHCAN', 'HTTPS', 'ENABLE_ACCESSLOG') - def __init__(self): - for k, v in self._DEFAULTS.items(): - setattr(self, k, os.environ.get(k, v)) +def load_settings() -> Settings: + configure_logging(os.environ.get("LOGLEVEL")) + try: + settings = Settings.from_env() + except SettingsError as exc: + log.error(str(exc)) + raise SystemExit(1) from exc + configure_logging(settings.LOGLEVEL) + return settings - for k, v in self.__dict__.items(): - if isinstance(v, str) and v.startswith('%%'): - setattr(self, k, getattr(self, v[2:])) - if k in self._BOOLEAN: - if v not in ('true', 'false', 'True', 'False', 'on', 'off', '1', '0'): - log.error(f'Environment variable "{k}" is set to a non-boolean value "{v}"') - sys.exit(1) - setattr(self, k, v in ('true', 'True', 'on', '1')) - if not self.URL_PREFIX.endswith('/'): - self.URL_PREFIX += '/' +Config = Settings - # Convert relative addresses to absolute addresses to prevent the failure of file address comparison - if self.YTDL_OPTIONS_FILE and self.YTDL_OPTIONS_FILE.startswith('.'): - self.YTDL_OPTIONS_FILE = str(Path(self.YTDL_OPTIONS_FILE).resolve()) - - self._runtime_overrides = {} - - success,_ = self.load_ytdl_options() - if not success: - sys.exit(1) - - def set_runtime_override(self, key, value): - self._runtime_overrides[key] = value - self.YTDL_OPTIONS[key] = value - - def remove_runtime_override(self, key): - self._runtime_overrides.pop(key, None) - self.YTDL_OPTIONS.pop(key, None) - - def _apply_runtime_overrides(self): - self.YTDL_OPTIONS.update(self._runtime_overrides) - - # Keys sent to the browser. Sensitive or server-only keys (YTDL_OPTIONS, - # paths, TLS config, etc.) are intentionally excluded. - _FRONTEND_KEYS = ( - 'CUSTOM_DIRS', - 'CREATE_CUSTOM_DIRS', - 'OUTPUT_TEMPLATE_CHAPTER', - 'PUBLIC_HOST_URL', - 'PUBLIC_HOST_AUDIO_URL', - 'DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT', - ) - - def frontend_safe(self) -> dict: - """Return only the config keys that are safe to expose to browser clients. - - Sensitive or server-only keys (YTDL_OPTIONS, file-system paths, TLS - settings, etc.) are intentionally excluded. - """ - return {k: getattr(self, k) for k in self._FRONTEND_KEYS} - - def load_ytdl_options(self) -> tuple[bool, str]: - try: - self.YTDL_OPTIONS = json.loads(os.environ.get('YTDL_OPTIONS', '{}')) - assert isinstance(self.YTDL_OPTIONS, dict) - except (json.decoder.JSONDecodeError, AssertionError): - msg = 'Environment variable YTDL_OPTIONS is invalid' - log.error(msg) - return (False, msg) - - if not self.YTDL_OPTIONS_FILE: - self._apply_runtime_overrides() - return (True, '') - - log.info(f'Loading yt-dlp custom options from "{self.YTDL_OPTIONS_FILE}"') - if not os.path.exists(self.YTDL_OPTIONS_FILE): - msg = f'File "{self.YTDL_OPTIONS_FILE}" not found' - log.error(msg) - return (False, msg) - try: - with open(self.YTDL_OPTIONS_FILE) as json_data: - opts = json.load(json_data) - assert isinstance(opts, dict) - except (json.decoder.JSONDecodeError, AssertionError): - msg = 'YTDL_OPTIONS_FILE contents is invalid' - log.error(msg) - return (False, msg) - - self.YTDL_OPTIONS.update(opts) - self._apply_runtime_overrides() - return (True, '') - -config = Config() -# Align root logger level with Config (keeps a single source of truth). -# This re-applies the log level after Config loads, in case LOGLEVEL was -# overridden by config file settings or differs from the environment variable. -logging.getLogger().setLevel(parseLogLevel(str(config.LOGLEVEL)) or logging.INFO) class ObjectSerializer(json.JSONEncoder): def default(self, obj): - # First try to use __dict__ for custom objects - if hasattr(obj, '__dict__'): + if hasattr(obj, "__dict__"): return obj.__dict__ - # Convert iterables (generators, dict_items, etc.) to lists - # Exclude strings and bytes which are also iterable - elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes)): + if hasattr(obj, "__iter__") and not isinstance(obj, (str, bytes)): try: return list(obj) except Exception: pass - # Fall back to default behavior return json.JSONEncoder.default(self, obj) + serializer = ObjectSerializer() -app = web.Application() -sio = socketio.AsyncServer(cors_allowed_origins='*') -sio.attach(app, socketio_path=config.URL_PREFIX + 'socket.io') -routes = web.RouteTableDef() -VALID_SUBTITLE_FORMATS = {'srt', 'txt', 'vtt', 'ttml', 'sbv', 'scc', 'dfxp'} -VALID_SUBTITLE_MODES = {'auto_only', 'manual_only', 'prefer_manual', 'prefer_auto'} -SUBTITLE_LANGUAGE_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9-]{0,34}$') -VALID_DOWNLOAD_TYPES = {'video', 'audio', 'captions', 'thumbnail'} -VALID_VIDEO_CODECS = {'auto', 'h264', 'h265', 'av1', 'vp9'} -VALID_VIDEO_FORMATS = {'any', 'mp4', 'ios'} -VALID_AUDIO_FORMATS = {'m4a', 'mp3', 'opus', 'wav', 'flac'} -VALID_THUMBNAIL_FORMATS = {'jpg'} + +SETTINGS_KEY = web.AppKey("settings", Settings) +SOCKETIO_KEY = web.AppKey("socketio", socketio.AsyncServer) +DQUEUE_KEY = web.AppKey("download_queue", DownloadQueue) +WATCH_TASK_KEY = web.AppKey("watch_task", asyncio.Task | None) +CUSTOM_DIRS_CACHE_KEY = web.AppKey("custom_dirs_cache", dict) + +VALID_SUBTITLE_FORMATS = {"srt", "txt", "vtt", "ttml", "sbv", "scc", "dfxp"} +VALID_SUBTITLE_MODES = {"auto_only", "manual_only", "prefer_manual", "prefer_auto"} +SUBTITLE_LANGUAGE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]{0,34}$") +VALID_DOWNLOAD_TYPES = {"video", "audio", "captions", "thumbnail"} +VALID_VIDEO_CODECS = {"auto", "h264", "h265", "av1", "vp9"} +VALID_VIDEO_FORMATS = {"any", "mp4", "ios"} +VALID_AUDIO_FORMATS = {"m4a", "mp3", "opus", "wav", "flac"} +VALID_THUMBNAIL_FORMATS = {"jpg"} + + +def _settings_from(request: web.Request) -> Settings: + return request.app[SETTINGS_KEY] + + +def _dqueue_from(request: web.Request) -> DownloadQueue: + return request.app[DQUEUE_KEY] + + +def _socketio_from(container: web.Request | web.Application) -> socketio.AsyncServer: + app = container.app if isinstance(container, web.Request) else container + return app[SOCKETIO_KEY] def _migrate_legacy_request(post: dict) -> dict: - """ - BACKWARD COMPATIBILITY: Translate old API request schema into the new one. - - Old API: - format (any/mp4/m4a/mp3/opus/wav/flac/thumbnail/captions) - quality - video_codec - subtitle_format (only when format=captions) - - New API: - download_type (video/audio/captions/thumbnail) - codec - format - quality - """ if "download_type" in post: return post @@ -229,14 +121,12 @@ def _migrate_legacy_request(post: dict) -> dict: post["format"] = str(post.get("subtitle_format") or "srt").strip().lower() post["quality"] = "best" else: - # old_format is usually any/mp4 (legacy video path) post["download_type"] = "video" post["codec"] = old_video_codec if old_quality == "best_ios": post["format"] = "ios" post["quality"] = "best" elif old_quality == "audio": - # Legacy "audio only" under video format maps to m4a audio. post["download_type"] = "audio" post["codec"] = "auto" post["format"] = "m4a" @@ -247,193 +137,281 @@ def _migrate_legacy_request(post: dict) -> dict: return post + class Notifier(DownloadQueueNotifier): + def __init__(self, app: web.Application): + self.app = app + async def added(self, dl): - log.info(f"Notifier: Download added - {dl.title}") - await sio.emit('added', serializer.encode(dl)) + log.info("Notifier: Download added - %s", dl.title) + await _socketio_from(self.app).emit("added", serializer.encode(dl)) async def updated(self, dl): - log.debug(f"Notifier: Download updated - {dl.title}") - await sio.emit('updated', serializer.encode(dl)) + log.debug("Notifier: Download updated - %s", dl.title) + await _socketio_from(self.app).emit("updated", serializer.encode(dl)) async def completed(self, dl): - log.info(f"Notifier: Download completed - {dl.title}") - await sio.emit('completed', serializer.encode(dl)) + log.info("Notifier: Download completed - %s", dl.title) + await _socketio_from(self.app).emit("completed", serializer.encode(dl)) - async def canceled(self, id): - log.info(f"Notifier: Download canceled - {id}") - await sio.emit('canceled', serializer.encode(id)) + async def canceled(self, identifier): + log.info("Notifier: Download canceled - %s", identifier) + await _socketio_from(self.app).emit("canceled", serializer.encode(identifier)) - async def cleared(self, id): - log.info(f"Notifier: Download cleared - {id}") - await sio.emit('cleared', serializer.encode(id)) + async def cleared(self, identifier): + log.info("Notifier: Download cleared - %s", identifier) + await _socketio_from(self.app).emit("cleared", serializer.encode(identifier)) -dqueue = DownloadQueue(config, Notifier()) -app.on_startup.append(lambda app: dqueue.initialize()) -app.on_cleanup.append(lambda app: Download.shutdown_manager()) class FileOpsFilter(DefaultFilter): + def __init__(self, settings: Settings): + super().__init__() + self.settings = settings + def __call__(self, change_type: int, path: str) -> bool: - # Check if this path matches our YTDL_OPTIONS_FILE - if path != config.YTDL_OPTIONS_FILE: + options_file = self.settings.YTDL_OPTIONS_FILE + if not options_file or path != options_file: return False - # For existing files, use samefile comparison to handle symlinks correctly - if os.path.exists(config.YTDL_OPTIONS_FILE): + if os.path.exists(options_file): try: - if not os.path.samefile(path, config.YTDL_OPTIONS_FILE): + if not os.path.samefile(path, options_file): return False except (OSError, IOError): - # If samefile fails, fall back to string comparison - if path != config.YTDL_OPTIONS_FILE: + if path != options_file: return False - - # Accept all change types for our file: modified, added, deleted return change_type in (Change.modified, Change.added, Change.deleted) -def get_options_update_time(success=True, msg=''): - result = { - 'success': success, - 'msg': msg, - 'update_time': None - } - # Only try to get file modification time if YTDL_OPTIONS_FILE is set and file exists - if config.YTDL_OPTIONS_FILE and os.path.exists(config.YTDL_OPTIONS_FILE): +def get_options_update_time(settings: Settings, success: bool = True, msg: str = "") -> dict[str, Any]: + result = {"success": success, "msg": msg, "update_time": None} + if settings.YTDL_OPTIONS_FILE and os.path.exists(settings.YTDL_OPTIONS_FILE): try: - result['update_time'] = os.path.getmtime(config.YTDL_OPTIONS_FILE) - except (OSError, IOError) as e: - log.warning(f"Could not get modification time for {config.YTDL_OPTIONS_FILE}: {e}") - result['update_time'] = None - + result["update_time"] = os.path.getmtime(settings.YTDL_OPTIONS_FILE) + except (OSError, IOError) as exc: + log.warning("Could not get modification time for %s: %s", settings.YTDL_OPTIONS_FILE, exc) return result -async def watch_files(): - async def _watch_files(): - async for changes in awatch(config.YTDL_OPTIONS_FILE, watch_filter=FileOpsFilter()): - success, msg = config.load_ytdl_options() - result = get_options_update_time(success, msg) - await sio.emit('ytdl_options_changed', serializer.encode(result)) - log.info(f'Starting Watch File: {config.YTDL_OPTIONS_FILE}') - asyncio.create_task(_watch_files()) +async def _watch_ytdl_options(app: web.Application) -> None: + settings = app[SETTINGS_KEY] + sio = _socketio_from(app) + log.info("Starting Watch File: %s", settings.YTDL_OPTIONS_FILE) + try: + async for _changes in awatch( + settings.YTDL_OPTIONS_FILE, + watch_filter=FileOpsFilter(settings), + ): + success, msg = settings.load_ytdl_options() + await sio.emit( + "ytdl_options_changed", + serializer.encode(get_options_update_time(settings, success, msg)), + ) + except asyncio.CancelledError: + raise -if config.YTDL_OPTIONS_FILE: - app.on_startup.append(lambda app: watch_files()) + +async def _initialize_app(app: web.Application) -> None: + settings = app[SETTINGS_KEY] + if settings.COOKIES_PATH.exists(): + settings.set_runtime_override("cookiefile", str(settings.COOKIES_PATH)) + log.info("Cookie file detected at %s", settings.COOKIES_PATH) + + await app[DQUEUE_KEY].initialize() + + if settings.YTDL_OPTIONS_FILE: + app[WATCH_TASK_KEY] = asyncio.create_task(_watch_ytdl_options(app)) + + +async def _cleanup_app(app: web.Application) -> None: + watch_task = app[WATCH_TASK_KEY] + if watch_task is not None: + watch_task.cancel() + with suppress(asyncio.CancelledError): + await watch_task + Download.shutdown_manager() async def _read_json_request(request: web.Request) -> dict: try: post = await request.json() except json.JSONDecodeError as exc: - raise web.HTTPBadRequest(reason='Invalid JSON request body') from exc + raise web.HTTPBadRequest(reason="Invalid JSON request body") from exc if not isinstance(post, dict): - raise web.HTTPBadRequest(reason='JSON request body must be an object') + raise web.HTTPBadRequest(reason="JSON request body must be an object") return post -@routes.post(config.URL_PREFIX + 'add') +def _normalize_origin(origin: str | None) -> str | None: + if not origin: + return None + try: + url = URL(origin) + except Exception: + return None + if not url.scheme or not url.host: + return None + try: + return str(url.origin()) + except ValueError: + return None + + +def _request_origin(request: web.Request) -> str | None: + try: + return str(request.url.origin()) + except ValueError: + origin = f"{request.scheme}://{request.host}" + return _normalize_origin(origin) + + +def _origin_allowed(request: web.Request, settings: Settings) -> str | None: + origin = _normalize_origin(request.headers.get("Origin")) + if origin is None: + return None + if origin == _request_origin(request): + return origin + if origin in settings.TRUSTED_ORIGINS: + return origin + return None + + async def add(request): + settings = _settings_from(request) + dqueue = _dqueue_from(request) + log.info("Received request to add download") post = await _read_json_request(request) post = _migrate_legacy_request(post) log.info( "Add download request: type=%s quality=%s format=%s has_folder=%s auto_start=%s", - post.get('download_type'), - post.get('quality'), - post.get('format'), - bool(post.get('folder')), - post.get('auto_start'), + post.get("download_type"), + post.get("quality"), + post.get("format"), + bool(post.get("folder")), + post.get("auto_start"), ) - url = post.get('url') - download_type = post.get('download_type') - codec = post.get('codec') - format = post.get('format') - quality = post.get('quality') + url = post.get("url") + download_type = post.get("download_type") + codec = post.get("codec") + format_value = post.get("format") + quality = post.get("quality") if not url or not quality or not download_type: log.error("Bad request: missing 'url', 'download_type', or 'quality'") raise web.HTTPBadRequest() - folder = post.get('folder') - custom_name_prefix = post.get('custom_name_prefix') - playlist_item_limit = post.get('playlist_item_limit') - auto_start = post.get('auto_start') - split_by_chapters = post.get('split_by_chapters') - chapter_template = post.get('chapter_template') - subtitle_language = post.get('subtitle_language') - subtitle_mode = post.get('subtitle_mode') + + folder = post.get("folder") + custom_name_prefix = post.get("custom_name_prefix") + playlist_item_limit = post.get("playlist_item_limit") + auto_start = post.get("auto_start") + split_by_chapters = post.get("split_by_chapters") + chapter_template = post.get("chapter_template") + subtitle_language = post.get("subtitle_language") + subtitle_mode = post.get("subtitle_mode") if custom_name_prefix is None: - custom_name_prefix = '' - if custom_name_prefix and ('..' in custom_name_prefix or custom_name_prefix.startswith('/') or custom_name_prefix.startswith('\\')): - raise web.HTTPBadRequest(reason='custom_name_prefix must not contain ".." or start with a path separator') + custom_name_prefix = "" + if custom_name_prefix and ( + ".." in custom_name_prefix + or custom_name_prefix.startswith("/") + or custom_name_prefix.startswith("\\") + ): + raise web.HTTPBadRequest( + reason='custom_name_prefix must not contain ".." or start with a path separator' + ) + if auto_start is None: auto_start = True if playlist_item_limit is None: - playlist_item_limit = config.DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT + playlist_item_limit = settings.DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT if split_by_chapters is None: split_by_chapters = False if chapter_template is None: - chapter_template = config.OUTPUT_TEMPLATE_CHAPTER + chapter_template = settings.OUTPUT_TEMPLATE_CHAPTER if subtitle_language is None: - subtitle_language = 'en' + subtitle_language = "en" if subtitle_mode is None: - subtitle_mode = 'prefer_manual' + subtitle_mode = "prefer_manual" + download_type = str(download_type).strip().lower() - codec = str(codec or 'auto').strip().lower() - format = str(format or '').strip().lower() + codec = str(codec or "auto").strip().lower() + format_value = str(format_value or "").strip().lower() quality = str(quality).strip().lower() subtitle_language = str(subtitle_language).strip() subtitle_mode = str(subtitle_mode).strip() - if chapter_template and ('..' in chapter_template or chapter_template.startswith('/') or chapter_template.startswith('\\')): - raise web.HTTPBadRequest(reason='chapter_template must not contain ".." or start with a path separator') + if chapter_template and ( + ".." in chapter_template + or chapter_template.startswith("/") + or chapter_template.startswith("\\") + ): + raise web.HTTPBadRequest( + reason='chapter_template must not contain ".." or start with a path separator' + ) if not SUBTITLE_LANGUAGE_RE.fullmatch(subtitle_language): - raise web.HTTPBadRequest(reason='subtitle_language must match pattern [A-Za-z0-9-] and be at most 35 characters') + raise web.HTTPBadRequest( + reason="subtitle_language must match pattern [A-Za-z0-9-] and be at most 35 characters" + ) if subtitle_mode not in VALID_SUBTITLE_MODES: - raise web.HTTPBadRequest(reason=f'subtitle_mode must be one of {sorted(VALID_SUBTITLE_MODES)}') - + raise web.HTTPBadRequest( + reason=f"subtitle_mode must be one of {sorted(VALID_SUBTITLE_MODES)}" + ) if download_type not in VALID_DOWNLOAD_TYPES: - raise web.HTTPBadRequest(reason=f'download_type must be one of {sorted(VALID_DOWNLOAD_TYPES)}') + raise web.HTTPBadRequest( + reason=f"download_type must be one of {sorted(VALID_DOWNLOAD_TYPES)}" + ) if codec not in VALID_VIDEO_CODECS: - raise web.HTTPBadRequest(reason=f'codec must be one of {sorted(VALID_VIDEO_CODECS)}') + raise web.HTTPBadRequest(reason=f"codec must be one of {sorted(VALID_VIDEO_CODECS)}") - if download_type == 'video': - if format not in VALID_VIDEO_FORMATS: - raise web.HTTPBadRequest(reason=f'format must be one of {sorted(VALID_VIDEO_FORMATS)} for video') - if quality not in {'best', 'worst', '2160', '1440', '1080', '720', '480', '360', '240'}: - raise web.HTTPBadRequest(reason="quality must be one of ['best', '2160', '1440', '1080', '720', '480', '360', '240', 'worst'] for video") - elif download_type == 'audio': - if format not in VALID_AUDIO_FORMATS: - raise web.HTTPBadRequest(reason=f'format must be one of {sorted(VALID_AUDIO_FORMATS)} for audio') - allowed_audio_qualities = {'best'} - if format == 'mp3': - allowed_audio_qualities |= {'320', '192', '128'} - elif format == 'm4a': - allowed_audio_qualities |= {'192', '128'} + if download_type == "video": + if format_value not in VALID_VIDEO_FORMATS: + raise web.HTTPBadRequest( + reason=f"format must be one of {sorted(VALID_VIDEO_FORMATS)} for video" + ) + if quality not in {"best", "worst", "2160", "1440", "1080", "720", "480", "360", "240"}: + raise web.HTTPBadRequest( + reason="quality must be one of ['best', '2160', '1440', '1080', '720', '480', '360', '240', 'worst'] for video" + ) + elif download_type == "audio": + if format_value not in VALID_AUDIO_FORMATS: + raise web.HTTPBadRequest( + reason=f"format must be one of {sorted(VALID_AUDIO_FORMATS)} for audio" + ) + allowed_audio_qualities = {"best"} + if format_value == "mp3": + allowed_audio_qualities |= {"320", "192", "128"} + elif format_value == "m4a": + allowed_audio_qualities |= {"192", "128"} if quality not in allowed_audio_qualities: - raise web.HTTPBadRequest(reason=f'quality must be one of {sorted(allowed_audio_qualities)} for format {format}') - codec = 'auto' - elif download_type == 'captions': - if format not in VALID_SUBTITLE_FORMATS: - raise web.HTTPBadRequest(reason=f'format must be one of {sorted(VALID_SUBTITLE_FORMATS)} for captions') - quality = 'best' - codec = 'auto' - elif download_type == 'thumbnail': - if format not in VALID_THUMBNAIL_FORMATS: - raise web.HTTPBadRequest(reason=f'format must be one of {sorted(VALID_THUMBNAIL_FORMATS)} for thumbnail') - quality = 'best' - codec = 'auto' + raise web.HTTPBadRequest( + reason=f"quality must be one of {sorted(allowed_audio_qualities)} for format {format_value}" + ) + codec = "auto" + elif download_type == "captions": + if format_value not in VALID_SUBTITLE_FORMATS: + raise web.HTTPBadRequest( + reason=f"format must be one of {sorted(VALID_SUBTITLE_FORMATS)} for captions" + ) + quality = "best" + codec = "auto" + elif download_type == "thumbnail": + if format_value not in VALID_THUMBNAIL_FORMATS: + raise web.HTTPBadRequest( + reason=f"format must be one of {sorted(VALID_THUMBNAIL_FORMATS)} for thumbnail" + ) + quality = "best" + codec = "auto" try: playlist_item_limit = int(playlist_item_limit) except (TypeError, ValueError) as exc: - raise web.HTTPBadRequest(reason='playlist_item_limit must be an integer') from exc + raise web.HTTPBadRequest(reason="playlist_item_limit must be an integer") from exc status = await dqueue.add( url, download_type, codec, - format, + format_value, quality, folder, custom_name_prefix, @@ -446,42 +424,48 @@ async def add(request): ) return web.Response(text=serializer.encode(status)) -@routes.post(config.URL_PREFIX + 'cancel-add') -async def cancel_add(request): - dqueue.cancel_add() - return web.Response(text=serializer.encode({'status': 'ok'}), content_type='application/json') -@routes.post(config.URL_PREFIX + 'delete') +async def cancel_add(request): + _dqueue_from(request).cancel_add() + return web.Response( + text=serializer.encode({"status": "ok"}), + content_type="application/json", + ) + + async def delete(request): + dqueue = _dqueue_from(request) post = await _read_json_request(request) - ids = post.get('ids') - where = post.get('where') - if not ids or where not in ['queue', 'done']: + ids = post.get("ids") + where = post.get("where") + if not ids or where not in ["queue", "done"]: log.error("Bad request: missing 'ids' or incorrect 'where' value") raise web.HTTPBadRequest() - status = await (dqueue.cancel(ids) if where == 'queue' else dqueue.clear(ids)) - log.info(f"Download delete request processed for ids: {ids}, where: {where}") + status = await (dqueue.cancel(ids) if where == "queue" else dqueue.clear(ids)) + log.info("Download delete request processed for ids: %s, where: %s", ids, where) return web.Response(text=serializer.encode(status)) -@routes.post(config.URL_PREFIX + 'start') + async def start(request): + dqueue = _dqueue_from(request) post = await _read_json_request(request) - ids = post.get('ids') - log.info(f"Received request to start pending downloads for ids: {ids}") + ids = post.get("ids") + log.info("Received request to start pending downloads for ids: %s", ids) status = await dqueue.start_pending(ids) return web.Response(text=serializer.encode(status)) -COOKIES_PATH = os.path.join(config.STATE_DIR, 'cookies.txt') - -@routes.post(config.URL_PREFIX + 'upload-cookies') async def upload_cookies(request): + settings = _settings_from(request) reader = await request.multipart() field = await reader.next() - if field is None or field.name != 'cookies': - return web.Response(status=400, text=serializer.encode({'status': 'error', 'msg': 'No cookies file provided'})) + if field is None or field.name != "cookies": + return web.Response( + status=400, + text=serializer.encode({"status": "error", "msg": "No cookies file provided"}), + ) - max_size = 1_000_000 # 1MB limit + max_size = 1_000_000 size = 0 content = bytearray() while True: @@ -490,197 +474,265 @@ async def upload_cookies(request): break size += len(chunk) if size > max_size: - return web.Response(status=400, text=serializer.encode({'status': 'error', 'msg': 'Cookie file too large (max 1MB)'})) + return web.Response( + status=400, + text=serializer.encode( + {"status": "error", "msg": "Cookie file too large (max 1MB)"} + ), + ) content.extend(chunk) - tmp_cookie_path = f"{COOKIES_PATH}.tmp" - with open(tmp_cookie_path, 'wb') as f: - f.write(content) - os.replace(tmp_cookie_path, COOKIES_PATH) - config.set_runtime_override('cookiefile', COOKIES_PATH) - log.info(f'Cookies file uploaded ({size} bytes)') - return web.Response(text=serializer.encode({'status': 'ok', 'msg': f'Cookies uploaded ({size} bytes)'})) + tmp_cookie_path = settings.COOKIES_PATH.with_suffix(".txt.tmp") + with tmp_cookie_path.open("wb") as handle: + handle.write(content) + os.replace(tmp_cookie_path, settings.COOKIES_PATH) + settings.set_runtime_override("cookiefile", str(settings.COOKIES_PATH)) + log.info("Cookies file uploaded (%s bytes)", size) + return web.Response( + text=serializer.encode({"status": "ok", "msg": f"Cookies uploaded ({size} bytes)"}) + ) + -@routes.post(config.URL_PREFIX + 'delete-cookies') async def delete_cookies(request): - has_uploaded_cookies = os.path.exists(COOKIES_PATH) - configured_cookiefile = config.YTDL_OPTIONS.get('cookiefile') - has_manual_cookiefile = isinstance(configured_cookiefile, str) and configured_cookiefile and configured_cookiefile != COOKIES_PATH + settings = _settings_from(request) + + has_uploaded_cookies = settings.COOKIES_PATH.exists() + configured_cookiefile = settings.YTDL_OPTIONS.get("cookiefile") + has_manual_cookiefile = ( + isinstance(configured_cookiefile, str) + and configured_cookiefile + and configured_cookiefile != str(settings.COOKIES_PATH) + ) if not has_uploaded_cookies: if has_manual_cookiefile: return web.Response( status=400, - text=serializer.encode({ - 'status': 'error', - 'msg': 'Cookies are configured manually via YTDL_OPTIONS (cookiefile). Remove or change that setting manually; UI delete only removes uploaded cookies.' - }) + text=serializer.encode( + { + "status": "error", + "msg": "Cookies are configured manually via YTDL_OPTIONS (cookiefile). Remove or change that setting manually; UI delete only removes uploaded cookies.", + } + ), ) - return web.Response(status=400, text=serializer.encode({'status': 'error', 'msg': 'No uploaded cookies to delete'})) + return web.Response( + status=400, + text=serializer.encode({"status": "error", "msg": "No uploaded cookies to delete"}), + ) - os.remove(COOKIES_PATH) - config.remove_runtime_override('cookiefile') - success, msg = config.load_ytdl_options() + settings.COOKIES_PATH.unlink() + settings.remove_runtime_override("cookiefile") + success, msg = settings.load_ytdl_options() if not success: - log.error(f'Cookies file deleted, but failed to reload YTDL_OPTIONS: {msg}') - return web.Response(status=500, text=serializer.encode({'status': 'error', 'msg': f'Cookies file deleted, but failed to reload YTDL_OPTIONS: {msg}'})) + log.error("Cookies file deleted, but failed to reload YTDL_OPTIONS: %s", msg) + return web.Response( + status=500, + text=serializer.encode( + { + "status": "error", + "msg": f"Cookies file deleted, but failed to reload YTDL_OPTIONS: {msg}", + } + ), + ) + + log.info("Cookies file deleted") + return web.Response(text=serializer.encode({"status": "ok"})) - log.info('Cookies file deleted') - return web.Response(text=serializer.encode({'status': 'ok'})) -@routes.get(config.URL_PREFIX + 'cookie-status') async def cookie_status(request): - configured_cookiefile = config.YTDL_OPTIONS.get('cookiefile') - has_configured_cookies = isinstance(configured_cookiefile, str) and os.path.exists(configured_cookiefile) - has_uploaded_cookies = os.path.exists(COOKIES_PATH) - exists = has_uploaded_cookies or has_configured_cookies - return web.Response(text=serializer.encode({'status': 'ok', 'has_cookies': exists})) + settings = _settings_from(request) + configured_cookiefile = settings.YTDL_OPTIONS.get("cookiefile") + has_configured_cookies = isinstance(configured_cookiefile, str) and os.path.exists( + configured_cookiefile + ) + exists = settings.COOKIES_PATH.exists() or has_configured_cookies + return web.Response(text=serializer.encode({"status": "ok", "has_cookies": exists})) + -@routes.get(config.URL_PREFIX + 'history') async def history(request): - history = { 'done': [], 'queue': [], 'pending': []} - - for _, v in dqueue.queue.saved_items(): - history['queue'].append(v) - for _, v in dqueue.done.saved_items(): - history['done'].append(v) - for _, v in dqueue.pending.saved_items(): - history['pending'].append(v) - + dqueue = _dqueue_from(request) + result = {"done": [], "queue": [], "pending": []} + for _, item in dqueue.queue.saved_items(): + result["queue"].append(item) + for _, item in dqueue.done.saved_items(): + result["done"].append(item) + for _, item in dqueue.pending.saved_items(): + result["pending"].append(item) log.info("Sending download history") - return web.Response(text=serializer.encode(history)) + return web.Response(text=serializer.encode(result)) -@sio.event -async def connect(sid, environ): - log.info(f"Client connected: {sid}") - await sio.emit('all', serializer.encode(dqueue.get()), to=sid) - await sio.emit('configuration', serializer.encode(config.frontend_safe()), to=sid) - if config.CUSTOM_DIRS: - await sio.emit('custom_dirs', serializer.encode(get_custom_dirs()), to=sid) - if config.YTDL_OPTIONS_FILE: - await sio.emit('ytdl_options_changed', serializer.encode(get_options_update_time()), to=sid) -def get_custom_dirs(): +def get_custom_dirs(app: web.Application): + settings = app[SETTINGS_KEY] + cache = app[CUSTOM_DIRS_CACHE_KEY] cache_ttl_seconds = 5 now = asyncio.get_running_loop().time() cache_key = ( - config.DOWNLOAD_DIR, - config.AUDIO_DOWNLOAD_DIR, - config.CUSTOM_DIRS_EXCLUDE_REGEX, + settings.DOWNLOAD_DIR, + settings.AUDIO_DOWNLOAD_DIR, + settings.CUSTOM_DIRS_EXCLUDE_REGEX, ) - if ( - hasattr(get_custom_dirs, "_cache_key") - and hasattr(get_custom_dirs, "_cache_value") - and hasattr(get_custom_dirs, "_cache_time") - and get_custom_dirs._cache_key == cache_key - and (now - get_custom_dirs._cache_time) < cache_ttl_seconds - ): - return get_custom_dirs._cache_value + cached = cache.get(cache_key) + if cached and (now - cached["time"]) < cache_ttl_seconds: + return cached["value"] - def recursive_dirs(base): + def recursive_dirs(base: str) -> list[str]: path = pathlib.Path(base) - # Converts PosixPath object to string, and remove base/ prefix - def convert(p): - s = str(p) - if s.startswith(base): - s = s[len(base):] + def convert(directory: pathlib.Path) -> str: + stringified = str(directory) + if stringified.startswith(base): + stringified = stringified[len(base) :] + if stringified.startswith("/"): + stringified = stringified[1:] + return stringified - if s.startswith('/'): - s = s[1:] - - return s - - # Include only directories which do not match the exclude filter - def include_dir(d): - if len(config.CUSTOM_DIRS_EXCLUDE_REGEX) == 0: + def include_dir(directory: str) -> bool: + if len(settings.CUSTOM_DIRS_EXCLUDE_REGEX) == 0: return True - else: - return re.search(config.CUSTOM_DIRS_EXCLUDE_REGEX, d) is None + return re.search(settings.CUSTOM_DIRS_EXCLUDE_REGEX, directory) is None - # Recursively lists all subdirectories of DOWNLOAD_DIR. - # Always include '' (the base directory itself) even when the - # directory is empty or does not yet exist. - dirs = list(filter(include_dir, map(convert, path.glob('**/')))) - if '' not in dirs: - dirs.insert(0, '') - - return dirs - - download_dir = recursive_dirs(config.DOWNLOAD_DIR) + directories = list(filter(include_dir, map(convert, path.glob("**/")))) + if "" not in directories: + directories.insert(0, "") + return directories + download_dir = recursive_dirs(settings.DOWNLOAD_DIR) audio_download_dir = download_dir - if config.DOWNLOAD_DIR != config.AUDIO_DOWNLOAD_DIR: - audio_download_dir = recursive_dirs(config.AUDIO_DOWNLOAD_DIR) + if settings.DOWNLOAD_DIR != settings.AUDIO_DOWNLOAD_DIR: + audio_download_dir = recursive_dirs(settings.AUDIO_DOWNLOAD_DIR) result = { "download_dir": download_dir, - "audio_download_dir": audio_download_dir + "audio_download_dir": audio_download_dir, } - get_custom_dirs._cache_key = cache_key - get_custom_dirs._cache_time = now - get_custom_dirs._cache_value = result + cache[cache_key] = {"time": now, "value": result} return result -@routes.get(config.URL_PREFIX) + async def index(request): - response = web.FileResponse(os.path.join(config.BASE_DIR, 'ui/dist/metube/browser/index.html')) - if 'metube_theme' not in request.cookies: - response.set_cookie('metube_theme', config.DEFAULT_THEME) + settings = _settings_from(request) + response = web.FileResponse(settings.UI_DIST_DIR / "index.html") + if "metube_theme" not in request.cookies: + response.set_cookie("metube_theme", settings.DEFAULT_THEME) return response -@routes.get(config.URL_PREFIX + 'robots.txt') + async def robots(request): - if config.ROBOTS_TXT: - response = web.FileResponse(os.path.join(config.BASE_DIR, config.ROBOTS_TXT)) - else: - response = web.Response( - text="User-agent: *\nDisallow: /download/\nDisallow: /audio_download/\n" - ) + settings = _settings_from(request) + if settings.ROBOTS_TXT_PATH: + return web.FileResponse(settings.ROBOTS_TXT_PATH) + return web.Response( + text="User-agent: *\nDisallow: /download/\nDisallow: /audio_download/\n" + ) + + +async def version(request): + return web.json_response({"yt-dlp": yt_dlp_version, "version": os.getenv("METUBE_VERSION", "dev")}) + + +async def index_redirect_root(request): + return web.HTTPFound(_settings_from(request).URL_PREFIX) + + +async def index_redirect_dir(request): + return web.HTTPFound(_settings_from(request).URL_PREFIX) + + +async def add_cors(request): + response = web.Response(text=serializer.encode({"status": "ok"})) + response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS" return response -@routes.get(config.URL_PREFIX + 'version') -async def version(request): - return web.json_response({ - "yt-dlp": yt_dlp_version, - "version": os.getenv("METUBE_VERSION", "dev") - }) - -if config.URL_PREFIX != '/': - @routes.get('/') - async def index_redirect_root(request): - return web.HTTPFound(config.URL_PREFIX) - - @routes.get(config.URL_PREFIX[:-1]) - async def index_redirect_dir(request): - return web.HTTPFound(config.URL_PREFIX) - -routes.static(config.URL_PREFIX + 'download/', config.DOWNLOAD_DIR, show_index=config.DOWNLOAD_DIRS_INDEXABLE) -routes.static(config.URL_PREFIX + 'audio_download/', config.AUDIO_DOWNLOAD_DIR, show_index=config.DOWNLOAD_DIRS_INDEXABLE) -routes.static(config.URL_PREFIX, os.path.join(config.BASE_DIR, 'ui/dist/metube/browser')) -try: - app.add_routes(routes) -except ValueError as e: - if 'ui/dist/metube/browser' in str(e): - raise RuntimeError('Could not find the frontend UI static assets. Please run `node_modules/.bin/ng build` inside the ui folder') from e - raise e - -# https://github.com/aio-libs/aiohttp/pull/4615 waiting for release -# @routes.options(config.URL_PREFIX + 'add') -async def add_cors(request): - return web.Response(text=serializer.encode({"status": "ok"})) - -app.router.add_route('OPTIONS', config.URL_PREFIX + 'add', add_cors) -app.router.add_route('OPTIONS', config.URL_PREFIX + 'cancel-add', add_cors) -app.router.add_route('OPTIONS', config.URL_PREFIX + 'upload-cookies', add_cors) -app.router.add_route('OPTIONS', config.URL_PREFIX + 'delete-cookies', add_cors) async def on_prepare(request, response): - if 'Origin' in request.headers: - response.headers['Access-Control-Allow-Origin'] = request.headers['Origin'] - response.headers['Access-Control-Allow-Headers'] = 'Content-Type' + origin = _origin_allowed(request, _settings_from(request)) + if origin is None: + return + response.headers["Access-Control-Allow-Origin"] = origin + response.headers["Access-Control-Allow-Headers"] = "Content-Type" + response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS" + response.headers["Vary"] = "Origin" + + +def register_socketio_handlers(app: web.Application, sio: socketio.AsyncServer) -> None: + @sio.event + async def connect(sid, environ): + settings = app[SETTINGS_KEY] + dqueue = app[DQUEUE_KEY] + log.info("Client connected: %s", sid) + await sio.emit("all", serializer.encode(dqueue.get()), to=sid) + await sio.emit("configuration", serializer.encode(settings.frontend_safe()), to=sid) + if settings.CUSTOM_DIRS: + await sio.emit("custom_dirs", serializer.encode(get_custom_dirs(app)), to=sid) + if settings.YTDL_OPTIONS_FILE: + await sio.emit( + "ytdl_options_changed", + serializer.encode(get_options_update_time(settings)), + to=sid, + ) + + +def register_routes(app: web.Application) -> None: + settings = app[SETTINGS_KEY] + prefix = settings.URL_PREFIX + + app.router.add_post(prefix + "add", add) + app.router.add_post(prefix + "cancel-add", cancel_add) + app.router.add_post(prefix + "delete", delete) + app.router.add_post(prefix + "start", start) + app.router.add_post(prefix + "upload-cookies", upload_cookies) + app.router.add_post(prefix + "delete-cookies", delete_cookies) + app.router.add_get(prefix + "cookie-status", cookie_status) + app.router.add_get(prefix + "history", history) + app.router.add_get(prefix, index) + app.router.add_get(prefix + "robots.txt", robots) + app.router.add_get(prefix + "version", version) + + if prefix != "/": + app.router.add_get("/", index_redirect_root) + app.router.add_get(prefix[:-1], index_redirect_dir) + + app.router.add_static( + prefix + "download/", + settings.DOWNLOAD_DIR, + show_index=settings.DOWNLOAD_DIRS_INDEXABLE, + ) + app.router.add_static( + prefix + "audio_download/", + settings.AUDIO_DOWNLOAD_DIR, + show_index=settings.DOWNLOAD_DIRS_INDEXABLE, + ) + app.router.add_static(prefix, str(settings.UI_DIST_DIR)) + + app.router.add_route("OPTIONS", prefix + "add", add_cors) + app.router.add_route("OPTIONS", prefix + "cancel-add", add_cors) + app.router.add_route("OPTIONS", prefix + "upload-cookies", add_cors) + app.router.add_route("OPTIONS", prefix + "delete-cookies", add_cors) + + +def create_app(settings: Settings | None = None) -> web.Application: + settings = settings or load_settings() + socket_cors_origins = list(settings.TRUSTED_ORIGINS) if settings.TRUSTED_ORIGINS else None + + app = web.Application() + app[SETTINGS_KEY] = settings + app[WATCH_TASK_KEY] = None + app[CUSTOM_DIRS_CACHE_KEY] = {} + + sio = socketio.AsyncServer(cors_allowed_origins=socket_cors_origins) + sio.attach(app, socketio_path=settings.URL_PREFIX + "socket.io") + app[SOCKETIO_KEY] = sio + app[DQUEUE_KEY] = DownloadQueue(settings, Notifier(app)) + + register_socketio_handlers(app, sio) + register_routes(app) + + app.on_startup.append(_initialize_app) + app.on_cleanup.append(_cleanup_app) + app.on_response_prepare.append(on_prepare) + return app -app.on_response_prepare.append(on_prepare) def supports_reuse_port(): try: @@ -691,25 +743,39 @@ def supports_reuse_port(): except (AttributeError, OSError): return False -def isAccessLogEnabled(): - if config.ENABLE_ACCESSLOG: + +def isAccessLogEnabled(settings: Settings): + if settings.ENABLE_ACCESSLOG: return access_logger - else: - return None - -if __name__ == '__main__': - logging.getLogger().setLevel(parseLogLevel(config.LOGLEVEL) or logging.INFO) - log.info(f"Listening on {config.HOST}:{config.PORT}") + return None - # Auto-detect cookie file on startup - if os.path.exists(COOKIES_PATH): - config.set_runtime_override('cookiefile', COOKIES_PATH) - log.info(f'Cookie file detected at {COOKIES_PATH}') +def main() -> None: + settings = load_settings() + application = create_app(settings) + log.info("Listening on %s:%s", settings.HOST, settings.PORT) - if config.HTTPS: + if settings.HTTPS: ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - ssl_context.load_cert_chain(certfile=config.CERTFILE, keyfile=config.KEYFILE) - web.run_app(app, host=config.HOST, port=int(config.PORT), reuse_port=supports_reuse_port(), ssl_context=ssl_context, access_log=isAccessLogEnabled()) - else: - web.run_app(app, host=config.HOST, port=int(config.PORT), reuse_port=supports_reuse_port(), access_log=isAccessLogEnabled()) + ssl_context.load_cert_chain(certfile=settings.CERTFILE, keyfile=settings.KEYFILE) + web.run_app( + application, + host=settings.HOST, + port=int(settings.PORT), + reuse_port=supports_reuse_port(), + ssl_context=ssl_context, + access_log=isAccessLogEnabled(settings), + ) + return + + web.run_app( + application, + host=settings.HOST, + port=int(settings.PORT), + reuse_port=supports_reuse_port(), + access_log=isAccessLogEnabled(settings), + ) + + +if __name__ == "__main__": + main() diff --git a/app/tests/conftest.py b/app/tests/conftest.py index 114806b..8de3b39 100644 --- a/app/tests/conftest.py +++ b/app/tests/conftest.py @@ -24,8 +24,8 @@ def _ensure_test_env() -> None: os.environ["TEMP_DIR"] = str(dl) os.environ["YTDL_OPTIONS"] = "{}" os.environ["YTDL_OPTIONS_FILE"] = "" - os.environ["BASE_DIR"] = str(base) os.environ["LOGLEVEL"] = "INFO" + os.environ["METUBE_TEST_APP_ROOT"] = str(base) os.environ["METUBE_TEST_ENV_READY"] = "1" diff --git a/app/tests/test_api.py b/app/tests/test_api.py index 4aa18e8..4d0ac32 100644 --- a/app/tests/test_api.py +++ b/app/tests/test_api.py @@ -1,33 +1,70 @@ -"""HTTP handler tests for ``main`` using mocked ``web.Request`` (no TestServer).""" +"""HTTP handler and app factory tests for ``main``.""" from __future__ import annotations import json +from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest from aiohttp import web +from yarl import URL import main +def _make_settings(tmp_path: Path, **overrides): + app_root = tmp_path / "app_root" + browser = app_root / "ui" / "dist" / "metube" / "browser" + browser.mkdir(parents=True) + (browser / "index.html").write_text("", encoding="utf-8") + + download_dir = tmp_path / "downloads" + state_dir = tmp_path / "state" + temp_dir = tmp_path / "temp" + download_dir.mkdir() + state_dir.mkdir() + temp_dir.mkdir() + + env = {k: str(v) for k, v in main.Config._DEFAULTS.items()} + env.update( + { + "DOWNLOAD_DIR": str(download_dir), + "AUDIO_DOWNLOAD_DIR": str(download_dir), + "STATE_DIR": str(state_dir), + "TEMP_DIR": str(temp_dir), + "YTDL_OPTIONS": "{}", + "YTDL_OPTIONS_FILE": "", + "LOGLEVEL": "INFO", + } + ) + env.update({key: str(value) for key, value in overrides.items()}) + return main.Config.from_env(env, app_root=app_root) + + @pytest.fixture -def mock_dqueue(monkeypatch): - d = MagicMock() - d.initialize = AsyncMock(return_value=None) - d.add = AsyncMock(return_value={"status": "ok"}) - d.cancel = AsyncMock(return_value={"status": "ok"}) - d.start_pending = AsyncMock(return_value={"status": "ok"}) - d.cancel_add = MagicMock() - d.queue = MagicMock() - d.done = MagicMock() - d.pending = MagicMock() - d.queue.saved_items = MagicMock(return_value=[]) - d.done.saved_items = MagicMock(return_value=[]) - d.pending.saved_items = MagicMock(return_value=[]) - d.get = MagicMock(return_value=([], [])) - monkeypatch.setattr(main, "dqueue", d) - return d +def app(tmp_path): + return main.create_app(_make_settings(tmp_path)) + + +@pytest.fixture +def mock_dqueue(app): + dqueue = MagicMock() + dqueue.initialize = AsyncMock(return_value=None) + dqueue.add = AsyncMock(return_value={"status": "ok"}) + dqueue.cancel = AsyncMock(return_value={"status": "ok"}) + dqueue.clear = AsyncMock(return_value={"status": "ok"}) + dqueue.start_pending = AsyncMock(return_value={"status": "ok"}) + dqueue.cancel_add = MagicMock() + dqueue.queue = MagicMock() + dqueue.done = MagicMock() + dqueue.pending = MagicMock() + dqueue.queue.saved_items = MagicMock(return_value=[]) + dqueue.done.saved_items = MagicMock(return_value=[]) + dqueue.pending.saved_items = MagicMock(return_value=[]) + dqueue.get = MagicMock(return_value=([], [])) + app[main.DQUEUE_KEY] = dqueue + return dqueue def _valid_video_add_body(**kwargs): @@ -42,48 +79,55 @@ def _valid_video_add_body(**kwargs): return base -def _json_request(body: dict | None): +def _request(app, body: dict | None = None): req = MagicMock(spec=web.Request) - req.json = AsyncMock(return_value=body) + req.app = app + req.headers = {} + req.cookies = {} + req.scheme = "http" + req.host = "localhost:8081" + req.url = URL("http://localhost:8081/") + if body is not None: + req.json = AsyncMock(return_value=body) return req @pytest.mark.asyncio -async def test_add_ok(mock_dqueue): - req = _json_request(_valid_video_add_body()) +async def test_add_ok(app, mock_dqueue): + req = _request(app, _valid_video_add_body()) resp = await main.add(req) assert resp.status == 200 - text = resp.text - data = json.loads(text) + data = json.loads(resp.text) assert data["status"] == "ok" mock_dqueue.add.assert_awaited_once() @pytest.mark.asyncio -async def test_add_missing_url_returns_400(mock_dqueue): - req = _json_request({"download_type": "video", "quality": "best", "format": "any"}) +async def test_add_missing_url_returns_400(app, mock_dqueue): + req = _request(app, {"download_type": "video", "quality": "best", "format": "any"}) with pytest.raises(web.HTTPBadRequest): await main.add(req) mock_dqueue.add.assert_not_called() @pytest.mark.asyncio -async def test_add_invalid_download_type(mock_dqueue): - req = _json_request(_valid_video_add_body(download_type="invalid")) +async def test_add_invalid_download_type(app, mock_dqueue): + req = _request(app, _valid_video_add_body(download_type="invalid")) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_add_invalid_video_quality(mock_dqueue): - req = _json_request(_valid_video_add_body(quality="9999")) +async def test_add_invalid_video_quality(app, mock_dqueue): + req = _request(app, _valid_video_add_body(quality="9999")) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_add_invalid_subtitle_language(mock_dqueue): - req = _json_request( +async def test_add_invalid_subtitle_language(app, mock_dqueue): + req = _request( + app, { "url": "https://example.com/v", "download_type": "captions", @@ -91,68 +135,66 @@ async def test_add_invalid_subtitle_language(mock_dqueue): "format": "srt", "quality": "best", "subtitle_language": "bad language!", - } + }, ) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_add_custom_name_prefix_path_traversal(mock_dqueue): - req = _json_request(_valid_video_add_body(custom_name_prefix="../evil")) +async def test_add_custom_name_prefix_path_traversal(app, mock_dqueue): + req = _request(app, _valid_video_add_body(custom_name_prefix="../evil")) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_add_chapter_template_path_traversal(mock_dqueue): - req = _json_request( +async def test_add_chapter_template_path_traversal(app, mock_dqueue): + req = _request( + app, _valid_video_add_body( split_by_chapters=True, chapter_template="/etc/passwd%(title)s", - ) + ), ) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_add_invalid_json_body(mock_dqueue): - req = MagicMock(spec=web.Request) +async def test_add_invalid_json_body(app, mock_dqueue): + req = _request(app) req.json = AsyncMock(side_effect=json.JSONDecodeError("msg", "", 0)) with pytest.raises(web.HTTPBadRequest): await main.add(req) @pytest.mark.asyncio -async def test_delete_missing_ids(mock_dqueue): - req = _json_request({"where": "queue"}) +async def test_delete_missing_ids(app, mock_dqueue): + req = _request(app, {"where": "queue"}) with pytest.raises(web.HTTPBadRequest): await main.delete(req) @pytest.mark.asyncio -async def test_delete_queue_calls_cancel(mock_dqueue): - req = _json_request({"where": "queue", "ids": ["http://x"]}) +async def test_delete_queue_calls_cancel(app, mock_dqueue): + req = _request(app, {"where": "queue", "ids": ["http://x"]}) resp = await main.delete(req) assert resp.status == 200 mock_dqueue.cancel.assert_awaited_once_with(["http://x"]) @pytest.mark.asyncio -async def test_start_pending(mock_dqueue): - req = _json_request({"ids": ["a"]}) +async def test_start_pending(app, mock_dqueue): + req = _request(app, {"ids": ["a"]}) resp = await main.start(req) assert resp.status == 200 mock_dqueue.start_pending.assert_awaited_once_with(["a"]) @pytest.mark.asyncio -async def test_history_shape(mock_dqueue): - mock_dqueue.queue.saved_items.return_value = [] - mock_dqueue.done.saved_items.return_value = [] - mock_dqueue.pending.saved_items.return_value = [] - req = MagicMock(spec=web.Request) +async def test_history_shape(app, mock_dqueue): + req = _request(app) resp = await main.history(req) assert resp.status == 200 data = json.loads(resp.text) @@ -160,8 +202,8 @@ async def test_history_shape(mock_dqueue): @pytest.mark.asyncio -async def test_version_json(mock_dqueue): - req = MagicMock(spec=web.Request) +async def test_version_json(app, mock_dqueue): + req = _request(app) resp = await main.version(req) assert resp.status == 200 body = json.loads(resp.text) @@ -169,8 +211,8 @@ async def test_version_json(mock_dqueue): @pytest.mark.asyncio -async def test_cookie_status(mock_dqueue): - req = MagicMock(spec=web.Request) +async def test_cookie_status(app, mock_dqueue): + req = _request(app) resp = await main.cookie_status(req) assert resp.status == 200 data = json.loads(resp.text) @@ -179,15 +221,15 @@ async def test_cookie_status(mock_dqueue): @pytest.mark.asyncio -async def test_options_add_cors(mock_dqueue): - req = MagicMock(spec=web.Request) +async def test_options_add_cors(app, mock_dqueue): + req = _request(app) resp = await main.add_cors(req) assert resp.status == 200 @pytest.mark.asyncio -async def test_upload_cookies_missing_field(mock_dqueue): - req = MagicMock(spec=web.Request) +async def test_upload_cookies_missing_field(app, mock_dqueue): + req = _request(app) reader = MagicMock() field = MagicMock() field.name = "wrongname" @@ -198,10 +240,47 @@ async def test_upload_cookies_missing_field(mock_dqueue): @pytest.mark.asyncio -async def test_add_legacy_format_migrated(mock_dqueue): - req = _json_request({"url": "https://example.com/v", "format": "m4a", "quality": "best"}) +async def test_add_legacy_format_migrated(app, mock_dqueue): + req = _request(app, {"url": "https://example.com/v", "format": "m4a", "quality": "best"}) resp = await main.add(req) assert resp.status == 200 call = mock_dqueue.add.await_args assert call is not None assert call.args[1] == "audio" + + +def test_create_app_registers_state(tmp_path): + app = main.create_app(_make_settings(tmp_path)) + assert app[main.SETTINGS_KEY].URL_PREFIX == "/" + assert main.DQUEUE_KEY in app + assert main.SOCKETIO_KEY in app + + +@pytest.mark.asyncio +async def test_on_prepare_allows_same_origin(app, mock_dqueue): + req = _request(app) + req.headers = {"Origin": "http://localhost:8081"} + response = web.Response() + await main.on_prepare(req, response) + assert response.headers["Access-Control-Allow-Origin"] == "http://localhost:8081" + + +@pytest.mark.asyncio +async def test_on_prepare_rejects_untrusted_origin(app, mock_dqueue): + req = _request(app) + req.headers = {"Origin": "https://evil.example"} + response = web.Response() + await main.on_prepare(req, response) + assert "Access-Control-Allow-Origin" not in response.headers + + +@pytest.mark.asyncio +async def test_on_prepare_allows_trusted_origin(tmp_path): + app = main.create_app( + _make_settings(tmp_path, TRUSTED_ORIGINS="https://trusted.example") + ) + req = _request(app) + req.headers = {"Origin": "https://trusted.example"} + response = web.Response() + await main.on_prepare(req, response) + assert response.headers["Access-Control-Allow-Origin"] == "https://trusted.example" diff --git a/app/tests/test_config.py b/app/tests/test_config.py index 0461ba1..6af860a 100644 --- a/app/tests/test_config.py +++ b/app/tests/test_config.py @@ -1,4 +1,4 @@ -"""Tests for ``Config`` (env parsing, yt-dlp options, frontend_safe).""" +"""Tests for ``Settings`` (env parsing, yt-dlp options, frontend_safe).""" from __future__ import annotations @@ -6,22 +6,37 @@ import json import os import tempfile import unittest +from pathlib import Path from unittest.mock import patch -from main import Config +from config import Settings, SettingsError + + +TEST_APP_ROOT = Path(os.environ["METUBE_TEST_APP_ROOT"]) def _base_env(**overrides: str) -> dict[str, str]: - env = {k: str(v) for k, v in Config._DEFAULTS.items()} + env = {k: str(v) for k, v in Settings._DEFAULTS.items()} + env.update( + { + "DOWNLOAD_DIR": os.environ["DOWNLOAD_DIR"], + "AUDIO_DOWNLOAD_DIR": os.environ["DOWNLOAD_DIR"], + "STATE_DIR": os.environ["STATE_DIR"], + "TEMP_DIR": os.environ["TEMP_DIR"], + "YTDL_OPTIONS": "{}", + "YTDL_OPTIONS_FILE": "", + "LOGLEVEL": "INFO", + } + ) env.update(overrides) return env class ConfigTests(unittest.TestCase): - def test_url_prefix_gets_trailing_slash(self): + def test_url_prefix_gets_normalized(self): with patch.dict(os.environ, _base_env(URL_PREFIX="foo"), clear=False): - c = Config() - self.assertEqual(c.URL_PREFIX, "foo/") + settings = Settings.from_env(app_root=TEST_APP_ROOT) + self.assertEqual(settings.URL_PREFIX, "/foo/") def test_ytdl_options_json_loaded(self): opts = {"quiet": True, "no_warnings": True} @@ -30,49 +45,69 @@ class ConfigTests(unittest.TestCase): _base_env(YTDL_OPTIONS=json.dumps(opts)), clear=False, ): - c = Config() - self.assertEqual(c.YTDL_OPTIONS["quiet"], True) + settings = Settings.from_env(app_root=TEST_APP_ROOT) + self.assertTrue(settings.YTDL_OPTIONS["quiet"]) - def test_invalid_ytdl_options_exits(self): + def test_invalid_ytdl_options_raises(self): with patch.dict(os.environ, _base_env(YTDL_OPTIONS="not-json"), clear=False): - with self.assertRaises(SystemExit): - Config() + with self.assertRaises(SettingsError): + Settings.from_env(app_root=TEST_APP_ROOT) - def test_invalid_boolean_env_exits(self): + def test_invalid_boolean_env_raises(self): with patch.dict(os.environ, _base_env(CUSTOM_DIRS="maybe"), clear=False): - with self.assertRaises(SystemExit): - Config() + with self.assertRaises(SettingsError): + Settings.from_env(app_root=TEST_APP_ROOT) def test_frontend_safe_excludes_secrets(self): with patch.dict(os.environ, _base_env(), clear=False): - c = Config() - safe = c.frontend_safe() + settings = Settings.from_env(app_root=TEST_APP_ROOT) + safe = settings.frontend_safe() self.assertNotIn("YTDL_OPTIONS", safe) self.assertNotIn("HOST", safe) def test_runtime_override_roundtrip(self): with patch.dict(os.environ, _base_env(), clear=False): - c = Config() - c.set_runtime_override("cookiefile", "/tmp/c.txt") - self.assertEqual(c.YTDL_OPTIONS.get("cookiefile"), "/tmp/c.txt") - c.remove_runtime_override("cookiefile") - self.assertIsNone(c.YTDL_OPTIONS.get("cookiefile")) + settings = Settings.from_env(app_root=TEST_APP_ROOT) + settings.set_runtime_override("cookiefile", "/tmp/c.txt") + self.assertEqual(settings.YTDL_OPTIONS.get("cookiefile"), "/tmp/c.txt") + settings.remove_runtime_override("cookiefile") + self.assertIsNone(settings.YTDL_OPTIONS.get("cookiefile")) def test_ytdl_options_file_merges(self): - with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: - json.dump({"extractor_args": {"youtube": {"player_client": ["web"]}}}, f) - path = f.name + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle: + json.dump({"extractor_args": {"youtube": {"player_client": ["web"]}}}, handle) + path = handle.name try: with patch.dict( os.environ, _base_env(YTDL_OPTIONS="{}", YTDL_OPTIONS_FILE=path), clear=False, ): - c = Config() - self.assertIn("extractor_args", c.YTDL_OPTIONS) + settings = Settings.from_env(app_root=TEST_APP_ROOT) + self.assertIn("extractor_args", settings.YTDL_OPTIONS) finally: os.unlink(path) + def test_missing_data_directory_raises(self): + missing = str(TEST_APP_ROOT / "missing-downloads") + with patch.dict(os.environ, _base_env(DOWNLOAD_DIR=missing), clear=False): + with self.assertRaises(SettingsError): + Settings.from_env(app_root=TEST_APP_ROOT) + + def test_https_requires_cert_and_key(self): + with patch.dict(os.environ, _base_env(HTTPS="true"), clear=False): + with self.assertRaises(SettingsError): + Settings.from_env(app_root=TEST_APP_ROOT) + + def test_sensitive_inline_ytdl_option_rejected(self): + with patch.dict( + os.environ, + _base_env(YTDL_OPTIONS=json.dumps({"cookiefile": "/tmp/cookies.txt"})), + clear=False, + ): + with self.assertRaises(SettingsError): + Settings.from_env(app_root=TEST_APP_ROOT) + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_entrypoint.py b/app/tests/test_entrypoint.py new file mode 100644 index 0000000..54d2819 --- /dev/null +++ b/app/tests/test_entrypoint.py @@ -0,0 +1,145 @@ +"""Tests for the hardened container entrypoint.""" + +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + +import pytest + + +REPO_ROOT = Path(__file__).resolve().parents[2] +ENTRYPOINT = REPO_ROOT / "docker-entrypoint.sh" + + +def _write_executable(path: Path, content: str) -> None: + path.write_text(content, encoding="utf-8") + path.chmod(0o755) + + +def _prepare_stub_bin(tmp_path: Path) -> Path: + stub_bin = tmp_path / "bin" + stub_bin.mkdir() + + _write_executable( + stub_bin / "id", + """#!/bin/sh +case "$1" in + -u) echo "${TEST_ID_U:-1000}" ;; + -g) echo "${TEST_ID_G:-1000}" ;; + *) exit 1 ;; +esac +""", + ) + _write_executable( + stub_bin / "gosu", + """#!/bin/sh +shift +"$@" +""", + ) + _write_executable( + stub_bin / "chown", + """#!/bin/sh +echo "$@" >> "${CHOWN_LOG}" +exit 0 +""", + ) + _write_executable( + stub_bin / "python3", + """#!/bin/sh +echo "$@" >> "${PYTHON_LOG}" +exit 0 +""", + ) + _write_executable( + stub_bin / "bgutil-pot", + """#!/bin/sh +echo "$@" >> "${BGUTIL_LOG}" +exit 0 +""", + ) + return stub_bin + + +def _base_env(tmp_path: Path) -> dict[str, str]: + download_dir = tmp_path / "downloads" + state_dir = tmp_path / "state" + temp_dir = tmp_path / "temp" + download_dir.mkdir(exist_ok=True) + state_dir.mkdir(exist_ok=True) + temp_dir.mkdir(exist_ok=True) + + return { + "DOWNLOAD_DIR": str(download_dir), + "STATE_DIR": str(state_dir), + "TEMP_DIR": str(temp_dir), + "PUID": "1000", + "PGID": "1000", + "UMASK": "022", + "CHOWN_DIRS": "true", + } + + +def _run_entrypoint(tmp_path: Path, **overrides: str) -> subprocess.CompletedProcess[str]: + stub_bin = _prepare_stub_bin(tmp_path) + env = os.environ.copy() + env.update(_base_env(tmp_path)) + env.update( + { + "PATH": f"{stub_bin}:{env['PATH']}", + "CHOWN_LOG": str(tmp_path / "chown.log"), + "PYTHON_LOG": str(tmp_path / "python.log"), + "BGUTIL_LOG": str(tmp_path / "bgutil.log"), + "TEST_ID_U": "0", + "TEST_ID_G": "0", + "UID": "", + "GID": "", + } + ) + env.update(overrides) + return subprocess.run( + ["sh", str(ENTRYPOINT)], + cwd=REPO_ROOT, + env=env, + text=True, + capture_output=True, + check=False, + ) + + +@pytest.mark.parametrize( + ("key", "value", "expected"), + [ + ("PUID", "abc", "PUID must be numeric"), + ("PGID", "abc", "PGID must be numeric"), + ("UMASK", "89", "UMASK must be a 3 or 4 digit octal value"), + ], +) +def test_entrypoint_rejects_invalid_identity_inputs(tmp_path, key, value, expected): + result = _run_entrypoint(tmp_path, **{key: value}) + assert result.returncode != 0 + assert expected in result.stderr + + +def test_entrypoint_chowns_only_data_directories(tmp_path): + result = _run_entrypoint(tmp_path) + assert result.returncode == 0 + chown_log = (tmp_path / "chown.log").read_text(encoding="utf-8") + assert "/app" not in chown_log + assert str(tmp_path / "downloads") in chown_log + assert str(tmp_path / "state") in chown_log + assert str(tmp_path / "temp") in chown_log + + +def test_entrypoint_fails_when_directories_are_not_accessible(tmp_path): + protected_dir = tmp_path / "state" + protected_dir.mkdir(exist_ok=True) + protected_dir.chmod(0) + try: + result = _run_entrypoint(tmp_path, CHOWN_DIRS="false") + finally: + protected_dir.chmod(0o700) + assert result.returncode != 0 + assert "Configured directories are not accessible" in result.stderr diff --git a/app/tests/test_main_helpers.py b/app/tests/test_main_helpers.py index 4258b79..326a9f4 100644 --- a/app/tests/test_main_helpers.py +++ b/app/tests/test_main_helpers.py @@ -4,10 +4,14 @@ from __future__ import annotations import json import logging +import os import unittest +from pathlib import Path import main +TEST_APP_ROOT = Path(os.environ["METUBE_TEST_APP_ROOT"]) + class MigrateLegacyRequestTests(unittest.TestCase): def test_already_new_schema_unchanged(self): @@ -94,7 +98,8 @@ class ObjectSerializerTests(unittest.TestCase): class FrontendSafeTests(unittest.TestCase): def test_only_expected_keys(self): - safe = main.config.frontend_safe() + settings = main.Config.from_env(app_root=TEST_APP_ROOT) + safe = settings.frontend_safe() for key in main.Config._FRONTEND_KEYS: self.assertIn(key, safe) self.assertNotIn("YTDL_OPTIONS", safe) diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 5f07bc5..6ac57ce 100644 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -1,28 +1,95 @@ #!/bin/sh +set -eu + +fail() { + echo "Error: $*" >&2 + exit 1 +} + +require_numeric() { + name="$1" + value="$2" + case "$value" in + ''|*[!0-9]*) + fail "${name} must be numeric" + ;; + esac +} + +require_octal_umask() { + case "$1" in + [0-7][0-7][0-7]|[0-7][0-7][0-7][0-7]) + ;; + *) + fail "UMASK must be a 3 or 4 digit octal value" + ;; + esac +} + +ensure_directory_access_current_user() { + path="$1" + label="$2" + [ -d "$path" ] || fail "${label} does not exist: ${path}" + [ -r "$path" ] || fail "${label} is not readable: ${path}" + [ -w "$path" ] || fail "${label} is not writable: ${path}" + [ -x "$path" ] || fail "${label} is not traversable: ${path}" +} + +ensure_directory_access_as_user() { + user_spec="$1" + shift + if ! gosu "$user_spec" sh -eu -c ' + for dir_path in "$@"; do + [ -d "$dir_path" ] || exit 10 + [ -r "$dir_path" ] || exit 11 + [ -w "$dir_path" ] || exit 12 + [ -x "$dir_path" ] || exit 13 + done + ' sh "$@"; then + fail "Configured directories are not accessible for ${user_spec}" + fi +} + +: "${PUID:=1000}" +: "${PGID:=1000}" +: "${UMASK:=022}" +: "${DOWNLOAD_DIR:?DOWNLOAD_DIR must be set}" +: "${STATE_DIR:?STATE_DIR must be set}" +: "${TEMP_DIR:?TEMP_DIR must be set}" + PUID="${UID:-$PUID}" PGID="${GID:-$PGID}" +require_numeric "PUID" "$PUID" +require_numeric "PGID" "$PGID" +require_octal_umask "$UMASK" + echo "Setting umask to ${UMASK}" -umask ${UMASK} +umask "$UMASK" + echo "Creating download directory (${DOWNLOAD_DIR}), state directory (${STATE_DIR}), and temp dir (${TEMP_DIR})" mkdir -p "${DOWNLOAD_DIR}" "${STATE_DIR}" "${TEMP_DIR}" -if [ `id -u` -eq 0 ] && [ `id -g` -eq 0 ]; then +if [ "$(id -u)" -eq 0 ] && [ "$(id -g)" -eq 0 ]; then if [ "${PUID}" -eq 0 ]; then echo "Warning: it is not recommended to run as root user, please check your setting of the PUID/PGID (or legacy UID/GID) environment variables" fi if [ "${CHOWN_DIRS:-true}" != "false" ]; then - echo "Changing ownership of download and state directories to ${PUID}:${PGID}" - chown -R "${PUID}":"${PGID}" /app "${DOWNLOAD_DIR}" "${STATE_DIR}" "${TEMP_DIR}" + echo "Changing ownership of data directories to ${PUID}:${PGID}" + chown -R "${PUID}:${PGID}" "${DOWNLOAD_DIR}" "${STATE_DIR}" "${TEMP_DIR}" fi + ensure_directory_access_as_user "${PUID}:${PGID}" "${DOWNLOAD_DIR}" "${STATE_DIR}" "${TEMP_DIR}" echo "Starting BgUtils POT Provider" - gosu "${PUID}":"${PGID}" bgutil-pot server >/tmp/bgutil-pot.log 2>&1 & + gosu "${PUID}:${PGID}" bgutil-pot server >/tmp/bgutil-pot.log 2>&1 & echo "Running MeTube as user ${PUID}:${PGID}" - exec gosu "${PUID}":"${PGID}" python3 app/main.py -else - echo "User set by docker; running MeTube as `id -u`:`id -g`" - echo "Starting BgUtils POT Provider" - bgutil-pot server >/tmp/bgutil-pot.log 2>&1 & - exec python3 app/main.py + exec gosu "${PUID}:${PGID}" python3 app/main.py fi + +echo "User set by docker; running MeTube as $(id -u):$(id -g)" +ensure_directory_access_current_user "${DOWNLOAD_DIR}" "DOWNLOAD_DIR" +ensure_directory_access_current_user "${STATE_DIR}" "STATE_DIR" +ensure_directory_access_current_user "${TEMP_DIR}" "TEMP_DIR" +echo "Starting BgUtils POT Provider" +bgutil-pot server >/tmp/bgutil-pot.log 2>&1 & +exec python3 app/main.py