mirror of
https://github.com/alexta69/metube.git
synced 2026-07-23 13:22:48 +00:00
fix: enforce download-dir containment at the resolved-path chokepoint
The per-download chapter_template was validated only against literal ".." in the template string, but yt-dlp expands %(section_title)s (and every other field) from attacker-controlled metadata at download time. On POSIX hosts yt-dlp does not neutralise a ".." path component, so a chapter titled ".." turns a guard-passing template like "%(section_title)s/%(section_title)s/x.%(ext)s" into "../../x.mp4" and writes outside DOWNLOAD_DIR. The same class of escape applies to any multi-segment output template (default/playlist/channel) whose fields resolve to "..". The template string can never see the "..": it only exists after expansion. So move the check to the one point every output path flows through — YoutubeDL.prepare_filename — via a _ConfinedYoutubeDL subclass that refuses any resolved path outside the download/temp roots (fail closed). This covers the main file, split-chapter files, thumbnails and subtitles in one place. With the chokepoint authoritative, the scattered ingress string checks (chapter_template, custom_name_prefix) and the weaker _output_dir_escapes literal-prefix check are removed. Tests move from the ingress layer to the chokepoint, exercising the real metadata-resolution vector. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -717,8 +717,6 @@ def parse_download_options(post: dict) -> dict:
|
||||
|
||||
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')
|
||||
if auto_start is None:
|
||||
auto_start = True
|
||||
if playlist_item_limit is None:
|
||||
@@ -743,8 +741,6 @@ def parse_download_options(post: dict) -> dict:
|
||||
enabled=config.ALLOW_YTDL_OPTIONS_OVERRIDES,
|
||||
)
|
||||
|
||||
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')
|
||||
if subtitle_mode not in VALID_SUBTITLE_MODES:
|
||||
|
||||
@@ -138,25 +138,6 @@ async def test_add_invalid_subtitle_language(mock_dqueue):
|
||||
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"))
|
||||
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(
|
||||
_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)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import pickle
|
||||
import signal
|
||||
import sys
|
||||
@@ -31,6 +32,25 @@ class _PostProcessor:
|
||||
self._downloader = downloader
|
||||
|
||||
|
||||
class _YoutubeDL:
|
||||
"""Minimal stand-in so ``_ConfinedYoutubeDL`` can subclass it under the shim.
|
||||
|
||||
``prepare_filename`` is patched per-test; the real containment logic lives in
|
||||
the ``_ConfinedYoutubeDL`` override, which is what the tests exercise.
|
||||
"""
|
||||
|
||||
def __init__(self, params=None, **kwargs):
|
||||
self.params = params or {}
|
||||
|
||||
def prepare_filename(self, *args, **kwargs):
|
||||
return ""
|
||||
|
||||
def add_post_processor(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
|
||||
fake_utils.DownloadError = type("DownloadError", (Exception,), {})
|
||||
fake_yt_dlp.YoutubeDL = _YoutubeDL
|
||||
fake_impersonate.ImpersonateTarget = _ImpersonateTarget
|
||||
fake_networking.impersonate = fake_impersonate
|
||||
fake_postprocessor_common.PostProcessor = _PostProcessor
|
||||
@@ -56,15 +76,17 @@ from ytdl import (
|
||||
_compact_persisted_entry,
|
||||
_convert_srt_to_txt_file,
|
||||
_AlbumArtistPostProcessor,
|
||||
_output_dir_escapes,
|
||||
_resolve_outtmpl_fields,
|
||||
_sanitize_entry_for_pickle,
|
||||
_sanitize_path_component,
|
||||
)
|
||||
|
||||
# Detect whether the real yt-dlp is loaded (as opposed to the minimal fake
|
||||
# shim above). _resolve_outtmpl_fields needs YoutubeDL at runtime.
|
||||
_has_real_ytdlp = hasattr(sys.modules.get("yt_dlp"), "YoutubeDL")
|
||||
# shim above). _resolve_outtmpl_fields needs YoutubeDL.evaluate_outtmpl at
|
||||
# runtime, which the shim's YoutubeDL stand-in deliberately does not provide.
|
||||
_has_real_ytdlp = hasattr(
|
||||
getattr(sys.modules.get("yt_dlp"), "YoutubeDL", None), "evaluate_outtmpl"
|
||||
)
|
||||
|
||||
|
||||
class AlbumArtistPostProcessorTests(unittest.TestCase):
|
||||
@@ -175,7 +197,7 @@ class AlbumArtistRegistrationTests(unittest.TestCase):
|
||||
download.info.download_type = 'audio'
|
||||
fake_ydl = MagicMock()
|
||||
|
||||
with patch('ytdl.yt_dlp.YoutubeDL', return_value=fake_ydl):
|
||||
with patch('ytdl._ConfinedYoutubeDL', return_value=fake_ydl):
|
||||
result = download._make_youtube_dl({'quiet': True})
|
||||
|
||||
self.assertIs(result, fake_ydl)
|
||||
@@ -187,7 +209,7 @@ class AlbumArtistRegistrationTests(unittest.TestCase):
|
||||
download = _make_test_download()
|
||||
fake_ydl = MagicMock()
|
||||
|
||||
with patch('ytdl.yt_dlp.YoutubeDL', return_value=fake_ydl):
|
||||
with patch('ytdl._ConfinedYoutubeDL', return_value=fake_ydl):
|
||||
download._make_youtube_dl({'quiet': True})
|
||||
|
||||
fake_ydl.add_post_processor.assert_not_called()
|
||||
@@ -301,18 +323,50 @@ class ResolveOuttmplFieldsTests(unittest.TestCase):
|
||||
self.assertFalse(literal_prefix.startswith('\\'))
|
||||
|
||||
|
||||
class OutputDirEscapesTests(unittest.TestCase):
|
||||
class ConfinedYoutubeDLTests(unittest.TestCase):
|
||||
"""The chokepoint: ``_ConfinedYoutubeDL.prepare_filename`` validates the
|
||||
*resolved* output path (after yt-dlp expands the template) and refuses any
|
||||
write outside the allowed roots. This is the single guard for the download-
|
||||
directory invariant across the main file, split-chapter files, thumbnails,
|
||||
subtitles, etc. — the ``..`` only exists post-expansion, so it is caught here
|
||||
rather than by any ingress string check.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.base_dir = tempfile.mkdtemp()
|
||||
self.base = os.path.realpath(tempfile.mkdtemp())
|
||||
|
||||
def test_relative_traversal_escapes(self):
|
||||
self.assertTrue(_output_dir_escapes(self.base_dir, '../../tmp/x/%(title)s.%(ext)s'))
|
||||
def _prepared_path(self, resolved, roots=None):
|
||||
ydl = ytdl._ConfinedYoutubeDL.__new__(ytdl._ConfinedYoutubeDL)
|
||||
ydl._allowed_roots = [self.base] if roots is None else roots
|
||||
with patch.object(
|
||||
ytdl.yt_dlp.YoutubeDL, "prepare_filename", return_value=resolved
|
||||
):
|
||||
return ydl.prepare_filename({})
|
||||
|
||||
def test_absolute_path_escapes(self):
|
||||
self.assertTrue(_output_dir_escapes(self.base_dir, '/tmp/x/%(title)s.%(ext)s'))
|
||||
def test_chapter_traversal_via_metadata_is_blocked(self):
|
||||
# e.g. chapter_template '%(section_title)s/%(section_title)s/pwned.%(ext)s'
|
||||
# with a chapter titled '..' expands to '../../pwned.mp4'.
|
||||
escaping = os.path.join(self.base, "..", "..", "pwned.mp4")
|
||||
with self.assertRaises(ytdl.yt_dlp.utils.DownloadError):
|
||||
self._prepared_path(escaping)
|
||||
|
||||
def test_normal_playlist_dir_stays_inside(self):
|
||||
self.assertFalse(_output_dir_escapes(self.base_dir, 'Playlist/%(title)s.%(ext)s'))
|
||||
def test_absolute_output_path_is_blocked(self):
|
||||
with self.assertRaises(ytdl.yt_dlp.utils.DownloadError):
|
||||
self._prepared_path("/etc/cron.d/evil")
|
||||
|
||||
def test_path_inside_download_dir_is_allowed(self):
|
||||
ok = os.path.join(self.base, "Playlist", "video.mp4")
|
||||
self.assertEqual(self._prepared_path(ok), ok)
|
||||
|
||||
def test_sibling_prefix_directory_is_blocked(self):
|
||||
# base '/x/downloads' must not be escapable to '/x/downloads-secret'.
|
||||
sibling = self.base + "-secret"
|
||||
with self.assertRaises(ytdl.yt_dlp.utils.DownloadError):
|
||||
self._prepared_path(os.path.join(sibling, "video.mp4"))
|
||||
|
||||
def test_empty_and_stdout_targets_pass_through(self):
|
||||
self.assertEqual(self._prepared_path(""), "")
|
||||
self.assertEqual(self._prepared_path("-"), "-")
|
||||
|
||||
|
||||
class SanitizeEntryForPickleTests(unittest.TestCase):
|
||||
|
||||
+34
-13
@@ -145,16 +145,36 @@ def _sanitize_path_component(value: Any) -> Any:
|
||||
return value.lstrip('.').strip() or '_'
|
||||
|
||||
|
||||
def _output_dir_escapes(base_dir: str, output_template: str) -> bool:
|
||||
"""True when the literal directory prefix of *output_template* resolves outside *base_dir*."""
|
||||
marker = output_template.find('%(')
|
||||
literal = output_template if marker == -1 else output_template[:marker]
|
||||
dir_prefix = os.path.dirname(literal)
|
||||
if not dir_prefix:
|
||||
return False
|
||||
real_base = os.path.realpath(base_dir)
|
||||
real_target = os.path.realpath(os.path.join(base_dir, dir_prefix))
|
||||
return not _is_within_directory(real_base, real_target)
|
||||
class _ConfinedYoutubeDL(yt_dlp.YoutubeDL):
|
||||
"""A ``YoutubeDL`` that refuses to emit any output path outside the allowed roots.
|
||||
|
||||
This is the single authoritative enforcement of MeTube's download-directory
|
||||
containment invariant. yt-dlp expands output templates at download time using
|
||||
metadata that is fully attacker-controlled (``%(title)s``, ``%(uploader)s``,
|
||||
``%(section_title)s`` from chapter titles, …) and, on POSIX hosts, does *not*
|
||||
neutralise a ``..`` path component — so any template segment resolving to
|
||||
``..`` next to a literal separator (or an absolute template) can traverse out
|
||||
of the download directory. Every output path — main file, split-chapter files,
|
||||
thumbnails, subtitles, infojson — is produced by ``prepare_filename``, so
|
||||
validating its result here covers them all, regardless of which template or
|
||||
metadata field carries the traversal. Checking the resolved path (rather than
|
||||
the template string) is what makes this robust: the ``..`` only exists after
|
||||
expansion, so no ingress string check can see it.
|
||||
"""
|
||||
|
||||
def __init__(self, params=None, *, allowed_roots=(), **kwargs):
|
||||
self._allowed_roots = [os.path.realpath(r) for r in allowed_roots if r]
|
||||
super().__init__(params=params, **kwargs)
|
||||
|
||||
def prepare_filename(self, *args, **kwargs):
|
||||
filename = super().prepare_filename(*args, **kwargs)
|
||||
if filename and filename != '-' and self._allowed_roots:
|
||||
resolved = os.path.realpath(filename)
|
||||
if not any(_is_within_directory(root, resolved) for root in self._allowed_roots):
|
||||
raise yt_dlp.utils.DownloadError(
|
||||
f'Refusing to write outside the download directory: {filename}'
|
||||
)
|
||||
return filename
|
||||
|
||||
|
||||
# Regex matching yt-dlp output-template field references, e.g. ``%(title)s``
|
||||
@@ -598,7 +618,10 @@ class Download:
|
||||
return put_status
|
||||
|
||||
def _make_youtube_dl(self, params):
|
||||
ydl = yt_dlp.YoutubeDL(params=params)
|
||||
ydl = _ConfinedYoutubeDL(
|
||||
params=params,
|
||||
allowed_roots=(self.download_dir, self.temp_dir),
|
||||
)
|
||||
if getattr(self.info, 'download_type', '') == 'audio':
|
||||
ydl.add_post_processor(_AlbumArtistPostProcessor(ydl), when='pre_process')
|
||||
return ydl
|
||||
@@ -1301,8 +1324,6 @@ class DownloadQueue:
|
||||
if playlist_item_limit > 0:
|
||||
log.info(f'playlist limit is set. Processing only first {playlist_item_limit} entries')
|
||||
ytdl_options['playlistend'] = playlist_item_limit
|
||||
if _output_dir_escapes(dldirectory, output):
|
||||
return {'status': 'error', 'msg': 'Refusing download: resolved output path escapes the download directory'}
|
||||
download = Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, dl.quality, dl.format, ytdl_options, dl)
|
||||
is_upcoming = (
|
||||
getattr(dl, 'live_status', None) == 'is_upcoming'
|
||||
|
||||
Reference in New Issue
Block a user