Merge 94c633d58a into 37cea84f77

* from https://github.com/yt-dlp/yt-dlp/security/advisories/GHSA-79w7-vh3h-8g4, thx grub4k

test/test_utils.py

@@ -14,9 +14,11 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 import io
 import itertools
 import json
+import types
 import xml.etree.ElementTree
 
 from youtube_dl.utils import (
+    _UnsafeExtensionError,
     age_restricted,
     args_to_str,
     base_url,
@@ -270,6 +272,27 @@ class TestUtil(unittest.TestCase):
             expand_path('~/%s' % env('YOUTUBE_DL_EXPATH_PATH')),
             '%s/expanded' % compat_getenv('HOME'))
 
+    _uncommon_extensions = [
+        ('exe', 'abc.exe.ext'),
+        ('de', 'abc.de.ext'),
+        ('../.mp4', None),
+        ('..\\.mp4', None),
+    ]
+
+    def assertUnsafeExtension(self, ext=None):
+        assert_raises = self.assertRaises(_UnsafeExtensionError)
+        assert_raises.ext = ext
+        orig_exit = assert_raises.__exit__
+
+        def my_exit(self_, exc_type, exc_val, exc_tb):
+            did_raise = orig_exit(exc_type, exc_val, exc_tb)
+            if did_raise and assert_raises.ext is not None:
+                self.assertEqual(assert_raises.ext, assert_raises.exception.extension, 'Unsafe extension  not as unexpected')
+            return did_raise
+
+        assert_raises.__exit__ = types.MethodType(my_exit, assert_raises)
+        return assert_raises
+
     def test_prepend_extension(self):
         self.assertEqual(prepend_extension('abc.ext', 'temp'), 'abc.temp.ext')
         self.assertEqual(prepend_extension('abc.ext', 'temp', 'ext'), 'abc.temp.ext')
@@ -278,6 +301,19 @@ class TestUtil(unittest.TestCase):
         self.assertEqual(prepend_extension('.abc', 'temp'), '.abc.temp')
         self.assertEqual(prepend_extension('.abc.ext', 'temp'), '.abc.temp.ext')
 
+        # Test uncommon extensions
+        self.assertEqual(prepend_extension('abc.ext', 'bin'), 'abc.bin.ext')
+        for ext, result in self._uncommon_extensions:
+            with self.assertUnsafeExtension(ext):
+                prepend_extension('abc', ext)
+            if result:
+                self.assertEqual(prepend_extension('abc.ext', ext, 'ext'), result)
+            else:
+                with self.assertUnsafeExtension(ext):
+                    prepend_extension('abc.ext', ext, 'ext')
+            with self.assertUnsafeExtension(ext):
+                prepend_extension('abc.unexpected_ext', ext, 'ext')
+
     def test_replace_extension(self):
         self.assertEqual(replace_extension('abc.ext', 'temp'), 'abc.temp')
         self.assertEqual(replace_extension('abc.ext', 'temp', 'ext'), 'abc.temp')
@@ -286,6 +322,16 @@ class TestUtil(unittest.TestCase):
         self.assertEqual(replace_extension('.abc', 'temp'), '.abc.temp')
         self.assertEqual(replace_extension('.abc.ext', 'temp'), '.abc.temp')
 
+        # Test uncommon extensions
+        self.assertEqual(replace_extension('abc.ext', 'bin'), 'abc.unknown_video')
+        for ext, _ in self._uncommon_extensions:
+            with self.assertUnsafeExtension(ext):
+                replace_extension('abc', ext)
+            with self.assertUnsafeExtension(ext):
+                replace_extension('abc.ext', ext, 'ext')
+            with self.assertUnsafeExtension(ext):
+                replace_extension('abc.unexpected_ext', ext, 'ext')
+
     def test_subtitles_filename(self):
         self.assertEqual(subtitles_filename('abc.ext', 'en', 'vtt'), 'abc.en.vtt')
         self.assertEqual(subtitles_filename('abc.ext', 'en', 'vtt', 'ext'), 'abc.en.vtt')

youtube_dl/YoutubeDL.py

@@ -7,6 +7,7 @@ import collections
 import copy
 import datetime
 import errno
+import functools
 import io
 import itertools
 import json
@@ -53,6 +54,7 @@ from .compat import (
     compat_urllib_request_DataHandler,
 )
 from .utils import (
+    _UnsafeExtensionError,
     age_restricted,
     args_to_str,
     bug_reports_message,
@@ -129,6 +131,20 @@ if compat_os_name == 'nt':
     import ctypes
 
 
+def _catch_unsafe_file_extension(func):
+    @functools.wraps(func)
+    def wrapper(self, *args, **kwargs):
+        try:
+            return func(self, *args, **kwargs)
+        except _UnsafeExtensionError as error:
+            self.report_error(
+                '{0} found; to avoid damaging your system, this value is disallowed.'
+                ' If you believe this is an error{1}').format(
+                    error.message, bug_reports_message(','))
+
+    return wrapper
+
+
 class YoutubeDL(object):
     """YoutubeDL class.
 
@@ -1925,6 +1941,7 @@ class YoutubeDL(object):
         if self.params.get('forcejson', False):
             self.to_stdout(json.dumps(self.sanitize_info(info_dict)))
 
+    @_catch_unsafe_file_extension
     def process_info(self, info_dict):
         """Process a single resolved IE result."""
 

youtube_dl/__init__.py

@@ -21,6 +21,7 @@ from .compat import (
     workaround_optparse_bug9161,
 )
 from .utils import (
+    _UnsafeExtensionError,
     DateRange,
     decodeOption,
     DEFAULT_OUTTMPL,
@@ -173,6 +174,9 @@ def _real_main(argv=None):
     if opts.ap_mso and opts.ap_mso not in MSO_INFO:
         parser.error('Unsupported TV Provider, use --ap-list-mso to get a list of supported TV Providers')
 
+    if opts.no_check_extensions:
+        _UnsafeExtensionError.lenient = True
+
     def parse_retries(retries):
         if retries in ('inf', 'infinite'):
             parsed_retries = float('inf')

youtube_dl/extractor/extractors.py

@@ -961,6 +961,7 @@ from .pluralsight import (
     PluralsightIE,
     PluralsightCourseIE,
 )
+from .podchaser import PodchaserIE
 from .podomatic import PodomaticIE
 from .pokemon import PokemonIE
 from .polskieradio import (

youtube_dl/extractor/podchaser.py

@@ -0,0 +1,114 @@
+# coding: utf-8
+from __future__ import unicode_literals
+import json
+import re
+from ..utils import float_or_none, try_get, str_to_int, unified_timestamp, merge_dicts
+from ..compat import compat_str
+from .common import InfoExtractor
+
+
+class PodchaserIE(InfoExtractor):
+    _VALID_URL = r'''(?x)
+        https?://(?:www\.)?podchaser\.com/
+        (?:
+            (?:podcasts/[\w-]+-(?P<podcast_id>[\d]+)))
+        (?:/episodes/[\w\-]+-
+            (?P<id>[\d]+))?'''
+
+    _TESTS = [{
+        'url': 'https://www.podchaser.com/podcasts/cum-town-36924/episodes/ep-285-freeze-me-off-104365585',
+        'info_dict': {
+            'id': '104365585',
+            'title': "Ep. 285 – freeze me off",
+            'description': 'cam ahn',
+            'thumbnail': r're:^https?://.*\.jpg$',
+            'ext': 'mp3',
+            'categories': ['Comedy'],
+            'tags': ['comedy', 'dark humor'],
+            'series': 'Cum Town',
+            'duration': 3708,
+            'timestamp': 1636531259,
+            'upload_date': '20211110'
+        }
+    }, {
+        'url': 'https://www.podchaser.com/podcasts/the-bone-zone-28853',
+        'info_dict': {
+            'id': '28853',
+            'title': 'The Bone Zone',
+            'description': 'Podcast by The Bone Zone',
+        },
+        'playlist_count': 275
+    }, {
+        'url': 'https://www.podchaser.com/podcasts/sean-carrolls-mindscape-scienc-699349/episodes',
+        'info_dict': {
+            'id': '699349',
+            'title': "Sean Carroll's Mindscape: Science, Society, Philosophy, Culture, Arts, and Ideas",
+            'description': 'md5:2cbd8f4749891a84dc8235342e0b5ff1'
+        },
+        'playlist_mincount': 199
+    }]
+
+    def _real_extract(self, url):
+        mobj = re.match(self._VALID_URL, url)
+        audio_id, podcast_id = mobj.group('id'), mobj.group('podcast_id')
+
+        # If one episode
+        if audio_id:
+            episodes = [self._download_json("https://api.podchaser.com/episodes/%s" % audio_id, audio_id)]
+
+        # Else get every episode available
+        else:
+            total_episode_count = self._download_json(
+                "https://api.podchaser.com/list/episode", podcast_id,
+                headers={'Content-Type': 'application/json;charset=utf-8'},
+                data=json.dumps({
+                    "filters": {"podcast_id": podcast_id}
+                }).encode()).get('total')
+            episodes = []
+            print(total_episode_count)
+            for i in range(total_episode_count // 100 + 1):
+                curr_episodes_data = self._download_json(
+                    "https://api.podchaser.com/list/episode", podcast_id,
+                    headers={'Content-Type': 'application/json;charset=utf-8'},
+                    data=json.dumps({
+                        "start": i * 100,
+                        "count": (i + 1) * 100,
+                        "sort_order": "SORT_ORDER_RECENT",
+                        "filters": {
+                            "podcast_id": podcast_id
+                        }, "options": {}
+                    }).encode())
+                curr_episodes = curr_episodes_data.get('entities') or []
+                if len(curr_episodes) + len(episodes) <= total_episode_count:
+                    episodes.extend(curr_episodes)
+
+        podcast_data = merge_dicts(
+            self._download_json("https://api.podchaser.com/podcasts/%s" % podcast_id, audio_id or podcast_id) or {},
+            episodes[0].get('podcast') or {} if episodes else {})
+
+        entries = [{
+            'id': compat_str(episode.get('id')),
+            'title': episode.get('title'),
+            'description': episode.get('description'),
+            'url': episode.get('audio_url'),
+            'thumbnail': episode.get('image_url'),
+            'duration': str_to_int(episode.get('length')),
+            'timestamp': unified_timestamp(episode.get('air_date')),
+            'rating': float_or_none(episode.get('rating')),
+            'categories': [
+                x.get('text') for x in
+                podcast_data.get('categories')
+                or try_get(podcast_data, lambda x: x['summary']['categories'], list) or []],
+            'tags': [tag.get('text') for tag in podcast_data.get('tags') or []],
+            'series': podcast_data.get('title'),
+        } for episode in episodes]
+
+        if len(entries) > 1:
+            # Return playlist
+            return self.playlist_result(
+                entries, playlist_id=compat_str(podcast_data.get('id')),
+                playlist_title=podcast_data.get('title'),
+                playlist_description=podcast_data.get('description'))
+
+        # Return episode
+        return entries[0]

youtube_dl/options.py

@@ -533,6 +533,10 @@ def parseOpts(overrideArguments=None):
         '--no-check-certificate',
         action='store_true', dest='no_check_certificate', default=False,
         help='Suppress HTTPS certificate validation')
+    workarounds.add_option(
+        '--no-check-extensions',
+        action='store_true', dest='no_check_extensions', default=False,
+        help='Suppress file extension validation')
     workarounds.add_option(
         '--prefer-insecure',
         '--prefer-unsecure', action='store_true', dest='prefer_insecure',

youtube_dl/utils.py

@@ -1717,21 +1717,6 @@ TIMEZONE_NAMES = {
     'PST': -8, 'PDT': -7   # Pacific
 }
 
-KNOWN_EXTENSIONS = (
-    'mp4', 'm4a', 'm4p', 'm4b', 'm4r', 'm4v', 'aac',
-    'flv', 'f4v', 'f4a', 'f4b',
-    'webm', 'ogg', 'ogv', 'oga', 'ogx', 'spx', 'opus',
-    'mkv', 'mka', 'mk3d',
-    'avi', 'divx',
-    'mov',
-    'asf', 'wmv', 'wma',
-    '3gp', '3g2',
-    'mp3',
-    'flac',
-    'ape',
-    'wav',
-    'f4f', 'f4m', 'm3u8', 'smil')
-
 # needed for sanitizing filenames in restricted mode
 ACCENT_CHARS = dict(zip('ÂÃÄÀÁÅÆÇÈÉÊËÌÍÎÏÐÑÒÓÔÕÖŐØŒÙÚÛÜŰÝÞßàáâãäåæçèéêëìíîïðñòóôõöőøœùúûüűýþÿ',
                         itertools.chain('AAAAAA', ['AE'], 'CEEEEIIIIDNOOOOOOO', ['OE'], 'UUUUUY', ['TH', 'ss'],
@@ -3959,19 +3944,22 @@ def parse_duration(s):
     return duration
 
 
-def prepend_extension(filename, ext, expected_real_ext=None):
+def _change_extension(prepend, filename, ext, expected_real_ext=None):
     name, real_ext = os.path.splitext(filename)
-    return (
-        '{0}.{1}{2}'.format(name, ext, real_ext)
-        if not expected_real_ext or real_ext[1:] == expected_real_ext
-        else '{0}.{1}'.format(filename, ext))
+    sanitize_extension = _UnsafeExtensionError.sanitize_extension
+
+    if not expected_real_ext or real_ext.partition('.')[0::2] == ('', expected_real_ext):
+        filename = name
+        if prepend and real_ext:
+            sanitize_extension(ext, prepend=prepend)
+            return ''.join((filename, '.', ext, real_ext))
+
+    # Mitigate path traversal and file impersonation attacks
+    return '.'.join((filename, sanitize_extension(ext)))
 
 
-def replace_extension(filename, ext, expected_real_ext=None):
-    name, real_ext = os.path.splitext(filename)
-    return '{0}.{1}'.format(
-        name if not expected_real_ext or real_ext[1:] == expected_real_ext else filename,
-        ext)
+prepend_extension = functools.partial(_change_extension, True)
+replace_extension = functools.partial(_change_extension, False)
 
 
 def check_executable(exe, args=[]):
@@ -6561,3 +6549,138 @@ def join_nonempty(*values, **kwargs):
     if from_dict is not None:
         values = (traverse_obj(from_dict, variadic(v)) for v in values)
     return delim.join(map(compat_str, filter(None, values)))
+
+
+class Namespace(object):
+    """Immutable namespace"""
+
+    def __init__(self, **kw_attr):
+        self.__dict__.update(kw_attr)
+
+    def __iter__(self):
+        return iter(self.__dict__.values())
+
+    @property
+    def items_(self):
+        return self.__dict__.items()
+
+
+MEDIA_EXTENSIONS = Namespace(
+    common_video=('avi', 'flv', 'mkv', 'mov', 'mp4', 'webm'),
+    video=('3g2', '3gp', 'f4v', 'mk3d', 'divx', 'mpg', 'ogv', 'm4v', 'wmv'),
+    common_audio=('aiff', 'alac', 'flac', 'm4a', 'mka', 'mp3', 'ogg', 'opus', 'wav'),
+    audio=('aac', 'ape', 'asf', 'f4a', 'f4b', 'm4b', 'm4p', 'm4r', 'oga', 'ogx', 'spx', 'vorbis', 'wma', 'weba'),
+    thumbnails=('jpg', 'png', 'webp'),
+    # storyboards=('mhtml', ),
+    subtitles=('srt', 'vtt', 'ass', 'lrc', 'ttml'),
+    manifests=('f4f', 'f4m', 'm3u8', 'smil', 'mpd'),
+)
+MEDIA_EXTENSIONS.video = MEDIA_EXTENSIONS.common_video + MEDIA_EXTENSIONS.video
+MEDIA_EXTENSIONS.audio = MEDIA_EXTENSIONS.common_audio + MEDIA_EXTENSIONS.audio
+
+KNOWN_EXTENSIONS = (
+    MEDIA_EXTENSIONS.video + MEDIA_EXTENSIONS.audio
+    + MEDIA_EXTENSIONS.manifests
+)
+
+
+class _UnsafeExtensionError(Exception):
+    """
+    Mitigation exception for unwanted file overwrite/path traversal
+
+    Ref: https://github.com/yt-dlp/yt-dlp/security/advisories/GHSA-79w7-vh3h-8g4j
+    """
+    _ALLOWED_EXTENSIONS = frozenset(itertools.chain(
+        (   # internal
+            'description',
+            'json',
+            'meta',
+            'orig',
+            'part',
+            'temp',
+            'uncut',
+            'unknown_video',
+            'ytdl',
+        ),
+        # video
+        MEDIA_EXTENSIONS.video, (
+            'avif',
+            'ismv',
+            'm2ts',
+            'm4s',
+            'mng',
+            'mpeg',
+            'qt',
+            'swf',
+            'ts',
+            'vp9',
+            'wvm',
+        ),
+        # audio
+        MEDIA_EXTENSIONS.audio, (
+            'isma',
+            'mid',
+            'mpga',
+            'ra',
+        ),
+        # image
+        MEDIA_EXTENSIONS.thumbnails, (
+            'bmp',
+            'gif',
+            'ico',
+            'heic',
+            'jng',
+            'jpeg',
+            'jxl',
+            'svg',
+            'tif',
+            'wbmp',
+        ),
+        # subtitle
+        MEDIA_EXTENSIONS.subtitles, (
+            'dfxp',
+            'fs',
+            'ismt',
+            'sami',
+            'scc',
+            'ssa',
+            'tt',
+        ),
+        # others
+        MEDIA_EXTENSIONS.manifests,
+        (
+            # not used in yt-dl
+            # *MEDIA_EXTENSIONS.storyboards,
+            # 'desktop',
+            # 'ism',
+            # 'm3u',
+            # 'sbv',
+            # 'swp',
+            # 'url',
+            # 'webloc',
+            # 'xml',
+        )))
+
+    def __init__(self, extension):
+        super(_UnsafeExtensionError, self).__init__('unsafe file extension: {0!r}'.format(extension))
+        self.extension = extension
+
+    # support --no-check-extensions
+    lenient = False
+
+    @classmethod
+    def sanitize_extension(cls, extension, **kwargs):
+        # ... /, *, prepend=False
+        prepend = kwargs.get('prepend', False)
+
+        if '/' in extension or '\\' in extension:
+            raise cls(extension)
+
+        if not prepend:
+            last = extension.rpartition('.')[-1]
+            if last == 'bin':
+                extension = last = 'unknown_video'
+            if not (cls.lenient or last.lower() in cls._ALLOWED_EXTENSIONS):
+                raise cls(extension)
+
+        return extension