From 63ea314b184c4a5fe0450510464609c5fc62b88e Mon Sep 17 00:00:00 2001 From: Media Date: Thu, 6 Aug 2026 19:23:02 -0500 Subject: [PATCH] build: harden controller v2 runtime --- .dockerignore | 21 +++ .env.example | 13 ++ Dockerfile | 62 ++++--- controller/config.py | 9 + controller/downloads.py | 166 ++++++++++++------ controller/mqtt_adapter.py | 4 +- controller/service.py | 77 ++++++++- docker-compose.yml | 59 +++++-- gunicorn.conf.py | 17 +- requirements.txt | 18 +- tests/test_config_contracts.py | 24 +++ tests/test_downloads.py | 79 ++++++++- tests/test_mqtt_adapter.py | 11 +- tests/test_runtime_assets.py | 302 +++++++++++++++++++++++++++++++++ tests/test_service.py | 59 ++++++- 15 files changed, 818 insertions(+), 103 deletions(-) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 tests/test_runtime_assets.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..59928e0 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,21 @@ +.git +.gitignore +.env +.env.* +!.env.example +__pycache__/ +*.py[cod] +.venv/ +venv/ +.pytest_cache/ +.coverage +htmlcov/ +*.sqlite3 +*.sqlite3-* +playlists.yaml +playlists.yaml.pre-v2.*.bak +controller-data/ +data/ +cookies*.txt +secrets/ +tests/ diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..b2f4744 --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +# Copy to .env, keep it private, and replace every example path/value. +FLASK_SECRET_KEY=generate-a-long-random-secret +CONTROLLER_DATA_DIR=/srv/video-player-controller/data +YTPLAYER_MEDIA_DIR=/mnt/Media/YTPlayer +CONTROLLER_SECRETS_DIR=/srv/video-player-controller/secrets + +MQTT_HOST=10.40.0.2 +MQTT_PORT=1883 +CONTROLLER_WEB_PORT=7531 + +# Match the owner of the host media and data directories. +CONTROLLER_UID=1000 +CONTROLLER_GID=1000 diff --git a/Dockerfile b/Dockerfile index 9f671de..32cb341 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,32 +1,48 @@ -FROM python:3.12-slim +# syntax=docker/dockerfile:1.7 -# System deps: ffmpeg for muxing, nodejs for yt-dlp JS runtime, ca-certs for TLS -RUN apt-get update && apt-get install -y --no-install-recommends \ - ffmpeg \ - nodejs \ - ca-certificates \ - && rm -rf /var/lib/apt/lists/* +# yt-dlp requires Node 22+ for current YouTube JavaScript challenges. Copy +# only the runtime binary into the smaller Python/ffmpeg application image. +FROM node:22.23.1-bookworm-slim AS javascript-runtime + +FROM python:3.12-slim-bookworm + +ARG CONTROLLER_UID=1000 +ARG CONTROLLER_GID=1000 + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 \ + PIP_NO_CACHE_DIR=1 \ + HOME=/tmp/controller-home \ + XDG_CACHE_HOME=/tmp/controller-cache \ + PORT=7531 + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates ffmpeg libatomic1 \ + && rm -rf /var/lib/apt/lists/* \ + && groupadd --gid "${CONTROLLER_GID}" controller \ + && useradd --uid "${CONTROLLER_UID}" --gid controller --create-home \ + --shell /usr/sbin/nologin controller \ + && install -d -o controller -g controller /app /controller-data /mnt/Media/YTPlayer + +COPY --from=javascript-runtime /usr/local/bin/node /usr/local/bin/node WORKDIR /app -COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt +COPY requirements.txt ./requirements.txt +RUN python -m pip install --requirement requirements.txt \ + && node --version \ + && python -c "from yt_dlp.utils._jsruntime import NodeJsRuntime; info = NodeJsRuntime().info; assert info and info.supported, info" \ + && ffmpeg -version -# Optional but strongly recommended: keep yt-dlp fresh (only if installed via pip) -RUN python -m pip install --no-cache-dir -U yt-dlp +COPY --chown=controller:controller controller ./controller +COPY --chown=controller:controller app.py controller_core.py wsgi.py gunicorn.conf.py config.yaml ./ -# Make yt-dlp defaults persistent (so your subprocess call doesn't need flags) -# 1) tell yt-dlp to use node as JS runtime -# 2) (optional) force android client to dodge SABR weirdness more often -RUN mkdir -p /etc/yt-dlp && printf "%s\n" \ - "--js-runtimes node" \ - "--extractor-args youtube:player_client=android" \ - > /etc/yt-dlp.conf +USER controller:controller -COPY app.py ./app.py -COPY controller_core.py ./controller_core.py +EXPOSE 7531 -ENV PYTHONUNBUFFERED=1 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:7531/healthz', timeout=3).read()"] -EXPOSE 8000 -CMD ["python", "app.py"] +CMD ["gunicorn", "--config", "gunicorn.conf.py", "wsgi:application"] diff --git a/controller/config.py b/controller/config.py index 2a69c1a..acced49 100644 --- a/controller/config.py +++ b/controller/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math import os from dataclasses import dataclass, field from pathlib import Path @@ -135,6 +136,8 @@ def _number(value: Any, name: str, *, minimum: float = 0.0) -> float: result = float(value) except (TypeError, ValueError, OverflowError) as exc: raise ConfigError(f"{name} must be a number") from exc + if not math.isfinite(result): + raise ConfigError(f"{name} must be finite") if result < minimum: raise ConfigError(f"{name} must be >= {minimum}") return result @@ -240,6 +243,10 @@ def load_config( raise ConfigError("storage.database_name must be a filename") if Path(storage.legacy_yaml_name).name != storage.legacy_yaml_name: raise ConfigError("storage.legacy_yaml_name must be a filename") + if not storage.media_base.is_absolute(): + raise ConfigError("storage.media_base must be an absolute path") + if not storage.data_base.is_absolute(): + raise ConfigError("storage.data_base must be an absolute path") cookie_raw = _env_value(env, "YTPLAYER_COOKIE_PATH", downloads_raw.get("cookie_path", "/run/secrets/yt-dlp-cookies.txt")) downloads = DownloadConfig( @@ -252,6 +259,8 @@ def load_config( ) if downloads.cookie_required and downloads.cookie_path is None: raise ConfigError("downloads.cookie_path is required when cookie_required is true") + if downloads.cookie_path is not None and not downloads.cookie_path.is_absolute(): + raise ConfigError("downloads.cookie_path must be an absolute path") unknown_behavior = _text(router_raw.get("when_state_unknown", "attempt"), "router.when_state_unknown") if unknown_behavior not in {"attempt", "continue", "reject"}: diff --git a/controller/downloads.py b/controller/downloads.py index 9ab02dd..ad0d8a0 100644 --- a/controller/downloads.py +++ b/controller/downloads.py @@ -7,9 +7,12 @@ import os import signal import shutil import subprocess +import tempfile import threading import time import uuid +from collections.abc import Iterator +from contextlib import contextmanager from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path @@ -171,37 +174,78 @@ class DownloadManager: except BaseException: LOG.exception("download event callback failed") - def _cookie_arguments(self) -> list[str]: - path = self.config.downloads.cookie_path - if path is None: - return [] - if path.is_file(): - return ["--cookies", str(path)] + def _working_cookie_copy(self) -> Path | None: + source = self.config.downloads.cookie_path + if source is None: + return None + if source.is_file(): + descriptor, temporary_name = tempfile.mkstemp( + prefix="ytplayer-cookies-", + suffix=".txt", + ) + temporary_path = Path(temporary_name) + try: + with os.fdopen(descriptor, "wb") as destination: + with source.open("rb") as source_file: + shutil.copyfileobj(source_file, destination) + destination.flush() + os.fsync(destination.fileno()) + temporary_path.chmod(0o600) + return temporary_path + except BaseException: + temporary_path.unlink(missing_ok=True) + raise if self.config.downloads.cookie_required: - raise FileNotFoundError(f"required yt-dlp cookie file is missing: {path}") - LOG.warning("yt-dlp cookie file is unavailable; continuing without it: %s", path) - return [] + raise FileNotFoundError( + f"required yt-dlp cookie file is missing: {source}" + ) + LOG.warning( + "yt-dlp cookie file is unavailable; continuing without it: %s", + source, + ) + return None - def _base_yt_dlp_arguments(self) -> list[str]: - return [self.config.downloads.yt_dlp_binary, "--ignore-config"] + self._cookie_arguments() + @contextmanager + def _yt_dlp_arguments(self) -> Iterator[list[str]]: + working_cookie = self._working_cookie_copy() + arguments = [ + self.config.downloads.yt_dlp_binary, + "--ignore-config", + "--no-js-runtimes", + "--js-runtimes", + "node", + "--cache-dir", + "/tmp/yt-dlp-cache", + ] + if working_cookie is not None: + arguments.extend(("--cookies", str(working_cookie))) + try: + yield arguments + finally: + if working_cookie is not None: + working_cookie.unlink(missing_ok=True) def _get_title(self, url: str) -> tuple[str, str | None]: - arguments = self._base_yt_dlp_arguments() + [ - "--flat-playlist", - "--print", - "%(playlist_title)s|||%(title)s|||%(id)s", - "--playlist-end", - "1", - url, - ] try: - result = self._run_command( - arguments, - check=True, - capture_output=True, - text=True, - timeout=min(120.0, self.config.downloads.process_timeout_seconds), - ) + with self._yt_dlp_arguments() as base_arguments: + arguments = base_arguments + [ + "--flat-playlist", + "--print", + "%(playlist_title)s|||%(title)s|||%(id)s", + "--playlist-end", + "1", + url, + ] + result = self._run_command( + arguments, + check=True, + capture_output=True, + text=True, + timeout=min( + 120.0, + self.config.downloads.process_timeout_seconds, + ), + ) line = (result.stdout or "").strip().splitlines()[0] playlist_title, _, remainder = line.partition("|||") video_title, _, yt_id = remainder.partition("|||") @@ -241,28 +285,29 @@ class DownloadManager: if slot_directory.is_symlink(): raise RuntimeError(f"refusing to download through symlink: {slot_directory}") title, yt_id = self._get_title(url) - arguments = self._base_yt_dlp_arguments() + [ - "--yes-playlist", - "--no-overwrites", - "--no-progress", - "-o", - str( - slot_directory - / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s" - ), - "-f", - ( - f"bv*[height<={self.config.downloads.max_height}]+ba/" - f"b[height<={self.config.downloads.max_height}]/b" - ), - url, - ] - self._run_command( - arguments, - check=True, - text=True, - timeout=self.config.downloads.process_timeout_seconds, - ) + with self._yt_dlp_arguments() as base_arguments: + arguments = base_arguments + [ + "--yes-playlist", + "--no-overwrites", + "--no-progress", + "-o", + str( + slot_directory + / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s" + ), + "-f", + ( + f"bv*[height<={self.config.downloads.max_height}]+ba/" + f"b[height<={self.config.downloads.max_height}]/b" + ), + url, + ] + self._run_command( + arguments, + check=True, + text=True, + timeout=self.config.downloads.process_timeout_seconds, + ) media_files = sorted( path for path in slot_directory.iterdir() @@ -467,6 +512,15 @@ class DownloadManager: recovered: list[int] = [] trash_root = self.config.storage.media_base / ".controller-trash" for token in self.store.pending_deletions(): + if any( + guard.slot == token.slot + for guard in self.store.list_live_guards() + ): + LOG.warning( + "deferring interrupted deletion for protected slot %s", + token.slot, + ) + continue if trash_root.is_dir(): prefix = f"slot{token.slot}.generation{token.generation}." for candidate in trash_root.iterdir(): @@ -479,6 +533,22 @@ class DownloadManager: recovered.append(token.slot) return recovered + def schedule_pending_deletion_recovery(self) -> bool: + """Run recoverable trash cleanup without delaying service startup.""" + + with self._lock: + if self._stopping: + return False + try: + future = self._executor.submit(self.recover_pending_deletions) + except BaseException: + LOG.exception("could not schedule interrupted deletion recovery") + return False + with self._lock: + self._futures.add(future) + future.add_done_callback(self._forget_future) + return True + def shutdown(self, *, wait: bool = True) -> None: with self._lock: self._stopping = True diff --git a/controller/mqtt_adapter.py b/controller/mqtt_adapter.py index 96f238c..8be0c6a 100644 --- a/controller/mqtt_adapter.py +++ b/controller/mqtt_adapter.py @@ -141,7 +141,9 @@ class MqttAdapter: self.config.mqtt.port, keepalive=self.config.mqtt.keepalive_seconds, ) - if result != mqtt.MQTT_ERR_SUCCESS: + # Paho 2.x returns None after configuring an asynchronous connect; + # older/test clients may return MQTT_ERR_SUCCESS. + if result not in (None, mqtt.MQTT_ERR_SUCCESS): raise RuntimeError(f"failed to configure MQTT connection: {result}") with self._lifecycle_lock: if ( diff --git a/controller/service.py b/controller/service.py index 6544f99..d8bf238 100644 --- a/controller/service.py +++ b/controller/service.py @@ -97,6 +97,9 @@ class ControllerService: self._next_expiry = 0.0 self._next_reconciliation = 0.0 self._processed_packets: dict[tuple[int, str, int], float] = {} + self._player_evidence_ready = False + self._operator_cleanup_override = False + self._recovery_scheduled = False @property def started(self) -> bool: @@ -144,7 +147,11 @@ class ControllerService: self.store, on_event=self._queue_download_event, ) - self.downloads.recover_pending_deletions() + # Cleanup is deliberately deferred until fresh Player evidence + # (or an explicit administrative reconciliation) makes it safe. + self._player_evidence_ready = False + self._operator_cleanup_override = False + self._recovery_scheduled = False self._stop_event.clear() self._next_expiry = ( self._monotonic() @@ -271,6 +278,8 @@ class ControllerService: degraded: list[str] = [] if started and not mqtt_connected: degraded.append("mqtt_disconnected") + if started and not self._cleanup_allowed(): + degraded.append("awaiting_player_evidence") if ( started and router_observation is not None @@ -375,6 +384,11 @@ class ControllerService: def _handle_call(self, action: str, payload: Any) -> Any: if action == "enqueue_download": assert self.downloads is not None + if not self._cleanup_allowed(): + raise ControllerUnavailableError( + "downloads are deferred until fresh Player evidence or " + "administrative playback reconciliation" + ) reservation = self.downloads.enqueue( payload["url"], payload.get("submitter"), payload.get("note") ) @@ -402,12 +416,16 @@ class ControllerService: handled_successfully = True return if event.kind == "connected": + self._player_evidence_ready = False + self._operator_cleanup_override = False self._announce_discovery() self._sync_slot_states(force=True) self._announce_reconciliation_reviews() handled_successfully = True return if event.kind == "disconnected": + self._player_evidence_ready = False + self._operator_cleanup_override = False assert self.playback is not None assert self.router is not None playback_effect = self.playback.mark_transport_uncertain( @@ -426,12 +444,27 @@ class ControllerService: payload = self._decode_json(event.payload, "Player status") if payload is not None: assert self.playback is not None + before_generation = self.store.player_observation().generation self._handle_playback_effects(self.playback.handle_status(payload)) + self._refresh_player_evidence(before_generation) elif event.topic == self.config.topics.player_telemetry: payload = self._decode_json(event.payload, "Player telemetry") if payload is not None: assert self.playback is not None + before_generation = self.store.player_observation().generation self._handle_playback_effects(self.playback.handle_telemetry(payload)) + self._refresh_player_evidence(before_generation) + elif event.topic == self.config.topics.player_state: + try: + flat_state = event.payload.decode("utf-8").strip() + except UnicodeDecodeError: + flat_state = "" + # Retained flat idle is sufficient only as a quiescence gate; + # it never mutates detailed playback state or clears guards. + self._player_evidence_ready = flat_state == "idle" + self._operator_cleanup_override = False + if self._player_evidence_ready: + self._schedule_deletion_recovery() elif event.topic == self.config.topics.controller_command: if event.retain: self._publish_event( @@ -478,8 +511,8 @@ class ControllerService: else: decision = self.router.observe_mode(mode) self._handle_router_decision(decision) - # Player flat state is intentionally only a Companion/cold-start - # hint. It is never reduced as authoritative playback truth. + # Flat state is only a cold-start cleanup gate. Detailed playback + # truth continues to come exclusively from status and telemetry. handled_successfully = True finally: if handled_successfully: @@ -828,6 +861,14 @@ class ControllerService: def _request_delete(self, payload: Mapping[str, Any]) -> dict[str, Any]: assert self.downloads is not None + if ( + not self._cleanup_allowed() + and payload.get("confirm_player_unknown") is not True + ): + raise SlotProtectedError( + "delete requires fresh Player evidence or " + "confirm_player_unknown=true" + ) slot = self._parse_slot(payload.get("slot")) expected_url = payload.get("url") if expected_url is not None and not isinstance(expected_url, str): @@ -886,6 +927,14 @@ class ControllerService: ) if not cleared: raise StaleOperationError("reconciliation review_id is stale") + observation = self.store.player_observation() + if ( + not self.store.list_live_guards() + and observation.active_slot is None + and observation.active_path is None + ): + self._operator_cleanup_override = True + self._schedule_deletion_recovery() assert self.router is not None self.router.abandon( self._session_lease_id, @@ -1118,6 +1167,26 @@ class ControllerService: if topic is not None: self.mqtt.publish_json(topic, payload, qos=1, retain=False) + def _cleanup_allowed(self) -> bool: + return self._player_evidence_ready or self._operator_cleanup_override + + def _schedule_deletion_recovery(self) -> None: + if self._recovery_scheduled or not self._cleanup_allowed(): + return + assert self.downloads is not None + self._recovery_scheduled = ( + self.downloads.schedule_pending_deletion_recovery() + ) + + def _refresh_player_evidence(self, before_generation: int) -> None: + observation = self.store.player_observation() + if observation.generation == before_generation: + return + self._operator_cleanup_override = False + self._player_evidence_ready = observation.availability == "online" + if self._player_evidence_ready and observation.phase == "idle": + self._schedule_deletion_recovery() + def _run_timers(self) -> None: assert self.playback is not None assert self.router is not None @@ -1133,7 +1202,7 @@ class ControllerService: if now >= self._next_reconciliation: self._handle_playback_effects(self.playback.mark_due_reconciliation()) self._next_reconciliation = now + 1.0 - if now >= self._next_expiry: + if now >= self._next_expiry and self._cleanup_allowed(): for slot in self.downloads.expire_due_slots(): self._announce_slot(slot, force=True) self._next_expiry = now + self.config.storage.expiry_interval_seconds diff --git a/docker-compose.yml b/docker-compose.yml index 0fb74c9..fc47c99 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,22 +1,61 @@ services: controller: - build: . + build: + context: . + args: + CONTROLLER_UID: ${CONTROLLER_UID:-1000} + CONTROLLER_GID: ${CONTROLLER_GID:-1000} + image: video-player-controller:local container_name: yt-controller restart: unless-stopped + init: true + read_only: true ports: - - "7531:7531" + - "${CONTROLLER_WEB_PORT:-7531}:7531" environment: - - FLASK_SECRET_KEY=change-me - - MQTT_HOST=10.40.0.2 - - MQTT_PORT=1883 + FLASK_SECRET_KEY: ${FLASK_SECRET_KEY:?Set FLASK_SECRET_KEY in .env} + MQTT_HOST: ${MQTT_HOST:-10.40.0.2} + MQTT_PORT: ${MQTT_PORT:-1883} + PORT: 7531 + YTPLAYER_DATA_BASE: /controller-data + YTPLAYER_MEDIA_BASE: /mnt/Media/YTPlayer + YTPLAYER_COOKIE_PATH: /run/secrets/yt-dlp-cookies.txt volumes: - - /mnt/Media:/mnt/Media # same path inside container - - /home/media/Scripts/Video-Player:/controller-data - - /home/media/Scripts/Video-Player/cookies.txt:/cookies_ro.txt:ro + - ${CONTROLLER_DATA_DIR:?Set CONTROLLER_DATA_DIR in .env}:/controller-data + - ${YTPLAYER_MEDIA_DIR:-/mnt/Media/YTPlayer}:/mnt/Media/YTPlayer + - ./config.yaml:/app/config.yaml:ro + # Optional: place yt-dlp-cookies.txt in this host directory. + - type: bind + source: ${CONTROLLER_SECRETS_DIR:-./secrets} + target: /run/secrets + read_only: true + tmpfs: + - /tmp:rw,nosuid,nodev,noexec,mode=1777,size=256m + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + healthcheck: + test: + - CMD + - python + - -c + - >- + import urllib.request; + urllib.request.urlopen('http://127.0.0.1:7531/healthz', timeout=3).read() + interval: 30s + timeout: 5s + start_period: 30s + retries: 3 + stop_grace_period: 45s + logging: + driver: json-file + options: + max-size: 10m + max-file: "3" networks: - tabernet - networks: tabernet: - external: true \ No newline at end of file + external: true diff --git a/gunicorn.conf.py b/gunicorn.conf.py index 6592bf4..fd4590e 100644 --- a/gunicorn.conf.py +++ b/gunicorn.conf.py @@ -2,10 +2,13 @@ from __future__ import annotations -import os +import sys + +from controller.config import load_config -bind = f"0.0.0.0:{int(os.environ.get('PORT', '7531'))}" +_web_config = load_config().web +bind = f"{_web_config.host}:{_web_config.port}" workers = 1 worker_class = "gthread" threads = 4 @@ -14,12 +17,14 @@ accesslog = "-" errorlog = "-" capture_output = True timeout = 30 -graceful_timeout = 15 +graceful_timeout = 30 +worker_tmp_dir = "/tmp" +control_socket_disable = True def worker_exit(server, worker): # pragma: no cover - exercised by Gunicorn - try: - from wsgi import service - except Exception: + module = sys.modules.get("wsgi") + service = getattr(module, "service", None) + if service is None: return service.stop() diff --git a/requirements.txt b/requirements.txt index f3c15d2..32365c5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,14 @@ -flask -paho-mqtt -yt-dlp -pyyaml +Flask==3.1.3 +gunicorn==26.0.0 +paho-mqtt==2.1.0 +PyYAML==6.0.3 +yt-dlp[default,pin]==2026.7.4 + +# Exact Flask/Gunicorn transitive set used by the production image. +blinker==1.9.0 +click==8.4.2 +itsdangerous==2.2.0 +Jinja2==3.1.6 +MarkupSafe==3.0.3 +packaging==26.3 +Werkzeug==3.1.8 diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py index b061202..d8f621a 100644 --- a/tests/test_config_contracts.py +++ b/tests/test_config_contracts.py @@ -115,6 +115,30 @@ class ConfigurationTests(unittest.TestCase): with self.assertRaisesRegex(ConfigError, "attempt, continue, or reject"): load_config(path, environ={}) + def test_non_finite_timeouts_are_rejected(self): + for value in (".nan", ".inf", "-.inf"): + with self.subTest(value=value), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + path.write_text( + f"router:\n confirmation_timeout_seconds: {value}\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ConfigError, "must be finite"): + load_config(path, environ={}) + + def test_runtime_paths_must_be_absolute(self): + cases = ( + ("storage:\n media_base: relative/media\n", "storage.media_base"), + ("storage:\n data_base: relative/data\n", "storage.data_base"), + ("downloads:\n cookie_path: relative/cookies.txt\n", "downloads.cookie_path"), + ) + for contents, field in cases: + with self.subTest(field=field), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + path.write_text(contents, encoding="utf-8") + with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"): + load_config(path, environ={}) + class ImportLifecycleTests(unittest.TestCase): def test_controller_core_import_does_not_connect_or_create_runtime_paths(self): diff --git a/tests/test_downloads.py b/tests/test_downloads.py index 859cf9e..feb7ecb 100644 --- a/tests/test_downloads.py +++ b/tests/test_downloads.py @@ -1,4 +1,5 @@ import os +import stat import subprocess import tempfile import threading @@ -24,14 +25,34 @@ class ImmediateExecutor: class FakeYtDlp: - def __init__(self, media_base: Path, *, fail_download=False, no_media=False): + def __init__( + self, + media_base: Path, + *, + fail_download=False, + no_media=False, + rewrite_cookie=False, + ): self.media_base = media_base self.fail_download = fail_download self.no_media = no_media + self.rewrite_cookie = rewrite_cookie self.calls = [] + self.cookie_observations = [] def __call__(self, arguments, **kwargs): self.calls.append((list(arguments), dict(kwargs))) + if "--cookies" in arguments: + cookie_path = Path(arguments[arguments.index("--cookies") + 1]) + self.cookie_observations.append( + ( + cookie_path, + cookie_path.read_text(encoding="utf-8"), + stat.S_IMODE(cookie_path.stat().st_mode), + ) + ) + if self.rewrite_cookie: + cookie_path.write_text("updated by yt-dlp", encoding="utf-8") if "--flat-playlist" in arguments: return subprocess.CompletedProcess( arguments, @@ -92,7 +113,8 @@ class DownloadManagerTests(unittest.TestCase): def test_success_uses_configured_cookie_and_atomically_builds_playlist(self): events = [] - runner = FakeYtDlp(self.media) + self.cookie.chmod(0o444) + runner = FakeYtDlp(self.media, rewrite_cookie=True) manager = self.manager(runner, events) result = manager.enqueue("https://example.test/playlist", "Alice", "Service") self.assertEqual(result.status, "reserved") @@ -109,20 +131,41 @@ class DownloadManagerTests(unittest.TestCase): self.assertEqual(list(playlist.parent.glob(".playlist.*.tmp")), []) for arguments, _ in runner.calls: self.assertIn("--ignore-config", arguments) + self.assertIn("--no-js-runtimes", arguments) + self.assertEqual(arguments[arguments.index("--js-runtimes") + 1], "node") + self.assertEqual( + arguments[arguments.index("--cache-dir") + 1], + "/tmp/yt-dlp-cache", + ) self.assertIn("--cookies", arguments) - self.assertIn(str(self.cookie), arguments) + self.assertNotIn(str(self.cookie), arguments) + self.assertEqual(len(runner.cookie_observations), 2) + working_paths = [item[0] for item in runner.cookie_observations] + self.assertEqual(len(set(working_paths)), 2) + self.assertTrue( + all(item[1] == "cookie data" for item in runner.cookie_observations) + ) + self.assertTrue( + all(item[2] == 0o600 for item in runner.cookie_observations) + ) + self.assertTrue(all(not path.exists() for path in working_paths)) + self.assertEqual(self.cookie.read_text(encoding="utf-8"), "cookie data") download_arguments = runner.calls[-1][0] self.assertIn("bv*[height<=720]+ba/b[height<=720]/b", download_arguments) def test_worker_failure_is_contained_and_persisted(self): events = [] - manager = self.manager(FakeYtDlp(self.media, fail_download=True), events) + runner = FakeYtDlp(self.media, fail_download=True) + manager = self.manager(runner, events) result = manager.enqueue("https://example.test/failure", None, None) self.assertEqual(result.status, "reserved") slot = self.store.get_slot(1) self.assertEqual(slot.media_state, "error") self.assertIn("exit code 7", slot.last_error) self.assertEqual(events[-1].state, "error") + self.assertTrue( + all(not observation[0].exists() for observation in runner.cookie_observations) + ) def test_missing_required_cookie_fails_job_without_escaping(self): self.cookie.unlink() @@ -196,6 +239,34 @@ class DownloadManagerTests(unittest.TestCase): self.assertFalse(slot.exists()) self.assertEqual(self.store.get_slot(1).media_state, "empty") + def test_interrupted_deletion_recovery_defers_a_new_playback_guard(self): + reservation = self.store.reserve_download( + "https://example.test/protected-recovery", + None, + None, + job_id="protected-recovery", + ) + self.store.fail_download( + "protected-recovery", + 1, + reservation.generation, + "failed", + ) + self.store.begin_delete(1, reason="manual") + slot = self.media / "slot1" + slot.mkdir(parents=True) + (slot / "still-playing.mp4").write_bytes(b"playing") + self.store.protect_slot(1, "player_active", guard_id="fresh-player") + manager = self.manager(FakeYtDlp(self.media)) + + self.assertEqual(manager.recover_pending_deletions(), []) + self.assertTrue(slot.exists()) + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + + self.store.clear_playback_guards(reason="fresh terminal idle evidence") + self.assertEqual(manager.recover_pending_deletions(), [1]) + self.assertFalse(slot.exists()) + def test_shutdown_terminates_tracked_process_without_waiting_for_job_timeout(self): manager = DownloadManager(self.config, self.store) errors = [] diff --git a/tests/test_mqtt_adapter.py b/tests/test_mqtt_adapter.py index a84e981..2e8ea69 100644 --- a/tests/test_mqtt_adapter.py +++ b/tests/test_mqtt_adapter.py @@ -32,6 +32,7 @@ class FakeClient: self.acks = [] self.suppress_exceptions = False self.connect_error = None + self.connect_result = None self.loop_result = mqtt.MQTT_ERR_SUCCESS self.publish_acknowledged = None self.publish_hook = None @@ -52,7 +53,7 @@ class FakeClient: self.calls.append(("connect_async", args, kwargs)) if self.connect_error is not None: raise self.connect_error - return mqtt.MQTT_ERR_SUCCESS + return self.connect_result def connect(self, *args, **kwargs): self.calls.append(("connect", args, kwargs)) @@ -123,6 +124,14 @@ class MqttAdapterTests(unittest.TestCase): self.assertEqual(connect[1], ("broker", 1884)) self.assertEqual(connect[2], {"keepalive": 30}) + def test_paho_two_connect_async_none_starts_network_loop(self): + self.client.connect_result = None + self.adapter.start() + + self.assertTrue(self.adapter.started) + self.assertIn("loop_start", [call[0] for call in self.client.calls]) + self.assertIsNone(self.adapter.last_error) + def test_initial_setup_oserror_does_not_escape_startup(self): self.client.connect_error = OSError("network unreachable") self.adapter.start() diff --git a/tests/test_runtime_assets.py b/tests/test_runtime_assets.py new file mode 100644 index 0000000..21ea4ae --- /dev/null +++ b/tests/test_runtime_assets.py @@ -0,0 +1,302 @@ +import ast +import re +import unittest +from pathlib import Path +from typing import Any + +import yaml + + +ROOT = Path(__file__).resolve().parents[1] + + +def _read(name: str) -> str: + return (ROOT / name).read_text(encoding="utf-8") + + +def _docker_logical_lines(text: str) -> list[str]: + logical: list[str] = [] + pending = "" + for raw_line in text.splitlines(): + stripped = raw_line.strip() + if not stripped or (not pending and stripped.startswith("#")): + continue + pending = f"{pending} {stripped}".strip() + if pending.endswith("\\"): + pending = pending[:-1].rstrip() + continue + logical.append(pending) + pending = "" + if pending: + logical.append(pending) + return logical + + +def _environment_map(value: Any) -> dict[str, str | None]: + if isinstance(value, dict): + return {str(key): None if item is None else str(item) for key, item in value.items()} + result: dict[str, str | None] = {} + for entry in value or []: + key, separator, item = str(entry).partition("=") + result[key] = item if separator else None + return result + + +def _volume_mounts(value: Any) -> list[tuple[str | None, str, bool]]: + mounts: list[tuple[str | None, str, bool]] = [] + for entry in value or []: + if isinstance(entry, dict): + target = str(entry.get("target") or "") + if target: + mounts.append( + ( + None if entry.get("source") is None else str(entry["source"]), + target, + bool(entry.get("read_only", False)), + ) + ) + continue + # Greedy source matching tolerates Compose substitutions containing + # `:-` or `:?` while keeping the absolute container target distinct. + match = re.match(r"^(.*):(/[^:]+)(?::([^:]+))?$", str(entry)) + if match: + mode = match.group(3) or "" + mounts.append((match.group(1), match.group(2), "ro" in mode.split(","))) + return mounts + + +def _secret_targets(value: Any) -> set[str]: + targets: set[str] = set() + for entry in value or []: + if isinstance(entry, dict): + source = str(entry.get("source") or "") + target = str(entry.get("target") or source) + else: + source = str(entry) + target = source + if not target: + continue + targets.add(target if target.startswith("/") else f"/run/secrets/{target}") + return targets + + +class RequirementPinTests(unittest.TestCase): + def test_runtime_requirements_are_exactly_pinned(self): + path = ROOT / "requirements.txt" + self.assertTrue(path.is_file(), "requirements.txt is missing") + requirements: dict[str, str] = {} + pattern = re.compile( + r"^([A-Za-z0-9_.-]+)(?:\[[A-Za-z0-9_,.-]+\])?==([^;\s]+)(?:\s*;.*)?$" + ) + for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + line = raw_line.split("#", 1)[0].strip() + if not line: + continue + match = pattern.fullmatch(line) + self.assertIsNotNone( + match, + f"requirements.txt:{line_number} is not an exact == pin: {line!r}", + ) + assert match is not None + normalized = re.sub(r"[-_.]+", "-", match.group(1)).lower() + requirements[normalized] = match.group(2) + + for project in ("flask", "gunicorn", "paho-mqtt", "pyyaml", "yt-dlp"): + self.assertIn(project, requirements, f"missing runtime dependency {project}") + self.assertNotIn("*", requirements[project]) + + +class DockerfileTests(unittest.TestCase): + def setUp(self): + self.path = ROOT / "Dockerfile" + self.assertTrue(self.path.is_file(), "Dockerfile is missing") + self.text = self.path.read_text(encoding="utf-8") + self.lines = _docker_logical_lines(self.text) + + def test_image_copies_the_v2_runtime(self): + copies = "\n".join( + line for line in self.lines if line.upper().startswith("COPY ") + ) + for required in ( + "controller", + "config.yaml", + "app.py", + "wsgi.py", + "gunicorn.conf.py", + ): + self.assertIn(required, copies, f"Dockerfile does not copy {required}") + + def test_image_runs_non_root_on_the_controller_port(self): + users = [line.split(None, 1)[1] for line in self.lines if line.upper().startswith("USER ")] + self.assertTrue(users, "Dockerfile must set a non-root USER") + self.assertNotIn(users[-1].strip().lower(), {"root", "0", "0:0"}) + self.assertRegex(self.text, r"(?im)^\s*EXPOSE\s+7531(?:/tcp)?\s*$") + + def test_image_has_an_http_healthcheck_and_runs_gunicorn(self): + self.assertRegex(self.text, r"(?im)^\s*HEALTHCHECK\b") + self.assertIn("/healthz", self.text) + launch = "\n".join( + line + for line in self.lines + if line.upper().startswith(("CMD ", "ENTRYPOINT ")) + ).lower() + self.assertIn("gunicorn", launch) + + def test_image_has_no_unbounded_yt_dlp_upgrade_or_hidden_global_config(self): + self.assertNotRegex( + self.text, + r"(?i)(?:--upgrade|-U)\s+[\"']?(?:yt-dlp|yt_dlp)", + ) + self.assertNotIn("/etc/yt-dlp.conf", self.text) + + def test_image_verifies_a_supported_node_runtime_for_yt_dlp(self): + self.assertRegex( + self.text, + r"(?im)^FROM\s+node:22\.\d+\.\d+-bookworm-slim\s+AS\s+", + ) + self.assertIn("NodeJsRuntime", self.text) + self.assertIn("info.supported", self.text) + + +class GunicornConfigTests(unittest.TestCase): + def test_gunicorn_uses_exactly_one_worker_without_preload(self): + path = ROOT / "gunicorn.conf.py" + self.assertTrue(path.is_file(), "gunicorn.conf.py is missing") + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + assignments: dict[str, Any] = {} + for node in tree.body: + if not isinstance(node, (ast.Assign, ast.AnnAssign)): + continue + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + try: + value = ast.literal_eval(node.value) + except (TypeError, ValueError): + continue + for target in targets: + if isinstance(target, ast.Name): + assignments[target.id] = value + + self.assertEqual(assignments.get("workers"), 1) + self.assertIs(assignments.get("preload_app"), False) + self.assertIs(assignments.get("control_socket_disable"), True) + self.assertEqual(assignments.get("worker_tmp_dir"), "/tmp") + + +class ComposeRuntimeTests(unittest.TestCase): + def setUp(self): + self.path = ROOT / "docker-compose.yml" + self.assertTrue(self.path.is_file(), "docker-compose.yml is missing") + self.text = self.path.read_text(encoding="utf-8") + document = yaml.safe_load(self.text) + self.assertIsInstance(document, dict) + self.document = document + self.service = document.get("services", {}).get("controller") + self.assertIsInstance(self.service, dict, "Compose controller service is missing") + + def test_compose_requires_a_real_flask_secret_and_publishes_port_7531(self): + environment = _environment_map(self.service.get("environment")) + secret = environment.get("FLASK_SECRET_KEY") + self.assertIsNotNone(secret, "FLASK_SECRET_KEY environment setting is missing") + assert secret is not None + self.assertTrue(secret.startswith("${FLASK_SECRET_KEY")) + self.assertIn("?", secret, "FLASK_SECRET_KEY must use required Compose substitution") + self.assertNotIn("change-me", secret.lower()) + self.assertNotIn("dev-secret", secret.lower()) + + published_targets: set[str] = set() + for entry in self.service.get("ports") or []: + if isinstance(entry, dict): + published_targets.add(str(entry.get("target"))) + else: + match = re.search(r":(\d+)(?:/(?:tcp|udp))?$", str(entry)) + if match: + published_targets.add(match.group(1)) + self.assertIn("7531", published_targets) + + def test_compose_uses_dedicated_data_media_and_cookie_mounts(self): + mounts = _volume_mounts(self.service.get("volumes")) + by_target = {target: (source, read_only) for source, target, read_only in mounts} + self.assertIn("/controller-data", by_target) + self.assertIn("/mnt/Media/YTPlayer", by_target) + self.assertNotEqual(by_target["/controller-data"][0], "/home/media/Scripts/Video-Player") + + cookie_path = "/run/secrets/yt-dlp-cookies.txt" + mounted_cookie = cookie_path in by_target and by_target[cookie_path][1] + mounted_secret_directory = ( + "/run/secrets" in by_target and by_target["/run/secrets"][1] + ) + secret_cookie = cookie_path in _secret_targets(self.service.get("secrets")) + self.assertTrue( + mounted_cookie or mounted_secret_directory or secret_cookie, + f"cookies must be read-only at exactly {cookie_path}", + ) + self.assertNotIn("/cookies_ro.txt", self.text) + self.assertNotRegex( + self.text, + r"/home/media/Scripts/Video-Player(?=[:\s\"'])", + ) + + def test_compose_enables_container_hardening_and_bounded_operations(self): + self.assertIs(self.service.get("read_only"), True) + self.assertIs(self.service.get("init"), True) + + cap_drop = self.service.get("cap_drop") or [] + if isinstance(cap_drop, str): + cap_drop = [cap_drop] + self.assertIn("ALL", {str(value).upper() for value in cap_drop}) + + security_options = self.service.get("security_opt") or [] + if isinstance(security_options, str): + security_options = [security_options] + self.assertTrue( + any("no-new-privileges" in str(value).lower() for value in security_options) + ) + + tmpfs = self.service.get("tmpfs") or [] + if isinstance(tmpfs, str): + tmpfs = [tmpfs] + self.assertTrue(any(str(value).split(":", 1)[0] == "/tmp" for value in tmpfs)) + self.assertTrue(any("noexec" in str(value) for value in tmpfs)) + + healthcheck = self.service.get("healthcheck") + self.assertIsInstance(healthcheck, dict) + self.assertIn("/healthz", str(healthcheck.get("test"))) + + logging_config = self.service.get("logging") + self.assertIsInstance(logging_config, dict) + options = logging_config.get("options") or {} + self.assertTrue(options.get("max-size"), "logging max-size is missing") + self.assertTrue(options.get("max-file"), "logging max-file is missing") + grace = str(self.service.get("stop_grace_period") or "").rstrip("s") + self.assertGreaterEqual(float(grace), 30.0) + + def test_compose_preserves_the_external_tabernet_network(self): + service_networks = self.service.get("networks") or [] + if isinstance(service_networks, dict): + service_networks = service_networks.keys() + self.assertIn("tabernet", service_networks) + tabernet = self.document.get("networks", {}).get("tabernet") + self.assertIsInstance(tabernet, dict) + self.assertIs(tabernet.get("external"), True) + + +class DockerIgnoreTests(unittest.TestCase): + def test_dockerignore_excludes_secrets_and_python_caches(self): + path = ROOT / ".dockerignore" + self.assertTrue(path.is_file(), ".dockerignore is missing") + patterns = { + line.strip().lower() + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + } + self.assertTrue(any("__pycache__" in pattern for pattern in patterns)) + self.assertTrue( + any("pyc" in pattern or "py[cod]" in pattern for pattern in patterns) + ) + self.assertTrue(any(pattern in {".env", ".env*", "*.env"} for pattern in patterns)) + self.assertTrue(any("secret" in pattern for pattern in patterns)) + self.assertTrue(any("cookie" in pattern for pattern in patterns)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 0b02953..f44e1b4 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -14,8 +14,9 @@ from controller.config import ( StorageConfig, ) from controller.downloads import canonical_playlist_path +from controller.models import SlotProtectedError from controller.mqtt_adapter import MqttEvent, PublishReceipt -from controller.service import ControllerService +from controller.service import ControllerService, ControllerUnavailableError from controller.store import ControllerStore @@ -82,12 +83,20 @@ class FakeMqttAdapter: class FakeDownloadManager: def __init__(self): self.recovered = False + self.recovery_calls = 0 + self.expiry_calls = 0 self.stopped = False def recover_pending_deletions(self): self.recovered = True + def schedule_pending_deletion_recovery(self): + self.recovered = True + self.recovery_calls += 1 + return True + def expire_due_slots(self): + self.expiry_calls += 1 return [] def shutdown(self, *, wait): @@ -144,7 +153,7 @@ class ControllerServiceTests(unittest.TestCase): wall_clock=self.wall, ) self.service.start() - self.assertTrue(self.downloads.recovered) + self.assertFalse(self.downloads.recovered) return self.service def ready_slot(self, slot=1): @@ -414,6 +423,52 @@ class ControllerServiceTests(unittest.TestCase): self.assertTrue(health["coordinator_alive"]) self.assertEqual(health["status"], "degraded") self.assertIn("mqtt_disconnected", health["degraded_reasons"]) + self.assertIn("awaiting_player_evidence", health["degraded_reasons"]) + + def test_cleanup_and_downloads_wait_for_post_start_player_evidence(self): + self.start_service() + self.service._next_expiry = self.monotonic() + time.sleep(0.2) + + self.assertEqual(self.downloads.expiry_calls, 0) + self.assertFalse(self.downloads.recovered) + with self.assertRaisesRegex( + ControllerUnavailableError, + "deferred until fresh Player evidence", + ): + self.service.enqueue_download( + "https://example.test/deferred", + None, + None, + ) + + self.send_message( + self.config.topics.player_state, + "idle", + retain=True, + ) + self.wait_for( + lambda: self.downloads.recovered, + "retained Player idle did not release deferred recovery", + ) + self.wait_for( + lambda: self.downloads.expiry_calls > 0, + "Player idle did not release deferred expiry", + ) + self.assertNotIn( + "awaiting_player_evidence", + self.service.health()["degraded_reasons"], + ) + + def test_manual_delete_requires_explicit_override_while_player_is_unknown(self): + self.start_service() + self.ready_slot(1) + + with self.assertRaisesRegex( + SlotProtectedError, + "confirm_player_unknown=true", + ): + self.service.submit_control({"action": "delete_slot", "slot": 1}) def test_administrator_can_clear_exact_uncertain_playback_review(self): self.start_service()