diff --git a/app/main.py b/app/main.py index fa575a9..f3fea3c 100644 --- a/app/main.py +++ b/app/main.py @@ -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: diff --git a/app/tests/test_api.py b/app/tests/test_api.py index 8ac86b3..fdf0588 100644 --- a/app/tests/test_api.py +++ b/app/tests/test_api.py @@ -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) diff --git a/app/tests/test_ytdl_utils.py b/app/tests/test_ytdl_utils.py index e3315a8..de13a7a 100644 --- a/app/tests/test_ytdl_utils.py +++ b/app/tests/test_ytdl_utils.py @@ -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): diff --git a/app/ytdl.py b/app/ytdl.py index a44e02b..f4f4553 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -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'