Repository: yt-dlp/yt-dlp
File: yt_dlp/extractor/subsplash.py (Python)

Code before:
import math

from .common import InfoExtractor
from ..utils import (
    InAdvancePagedList,
    get_element_by_id,
    int_or_none,
    parse_iso8601,
    try_call,
    url_or_none,
)
from ..utils.traversal import traverse_obj


class SubsplashBaseIE(InfoExtractor):
    def _get_headers(self, url, display_id):
        token = try_call(lambda: self._get_cookies(url)['ss-token-guest'].value)
        if not token:
            webpage, urlh = self._download_webpage_handle(url, display_id)
            token = try_call(lambda: self._get_cookies(url)['ss-token-guest'].value)
            if not token:
                token = urlh.get_header('x-api-token')
            if not token:
                token = traverse_obj(get_element_by_id('shoebox-tokens', webpage), ({json.loads}, 'apiToken', {str}))
            if not token:
                token = self._search_regex(r'\\"tokens\\":{\\"guest\\":\\"([A-Za-z0-9._-]+)\\"', webpage, 'token', default=None)

        if not token:
            return None
        return {'Authorization': f'Bearer {token}'}

    def _extract_video(self, data, display_id):
        formats = []
        video_data = traverse_obj(data, ('_embedded', 'video', '_embedded', {dict}))
        m3u8_url = traverse_obj(video_data, ('playlists', 0, '_links', 'related', 'href', {url_or_none}))
        if m3u8_url:
            formats.extend(self._extract_m3u8_formats(m3u8_url, display_id, 'mp4', m3u8_id='hls', fatal=False))
        mp4_entry = traverse_obj(video_data, ('video-outputs', lambda _, v: url_or_none(v['_links']['related']['href']), any))
        if mp4_entry:
            formats.append({
                'url': mp4_entry['_links']['related']['href'],
                'format_id': 'direct',
                'preference': 1,
                **traverse_obj(mp4_entry, {
                    'height': ('height', {int_or_none}),
                    'width': ('width', {int_or_none}),
                    'filesize': ('file_size', {int_or_none}),
                }),
            })
        return {
            'formats': formats,

Review comment (security):
```suggestion
            token = (
                try_call(lambda: self._get_cookies(url)['ss-token-guest'].value)
                or urlh.get_header('x-api-token')
                or traverse_obj(webpage, ({find_element(id='shoebox-tokens')}, {json.loads}, 'apiToken', {str}))
                or self._search_regex(r'\\"tokens\\":{\\"guest\\":\\"([A-Za-z0-9._-]+)\\"', webpage, 'token', default=None))
```

A couple of additional points, though:

- It's usually bad practice to use the `get_element` or `find_element` utils when you need to find/parse JSON data... these utils unescape all HTML special chars which could cause JSON parsing errors. It's almost always better to use `_search_json()` instead

- If the token isn't being set as a cookie, should we cache it to a class variable?

Code after:

from .common import InfoExtractor
from ..utils import (
    InAdvancePagedList,
    int_or_none,
    parse_iso8601,
    try_call,
    url_or_none,
)
from ..utils.traversal import traverse_obj


class SubsplashBaseIE(InfoExtractor):
    def _get_headers(self, url, display_id):
        token = try_call(lambda: self._get_cookies(url)['ss-token-guest'].value)
        if not token:
            webpage, urlh = self._download_webpage_handle(url, display_id)
            token = (
                try_call(lambda: self._get_cookies(url)['ss-token-guest'].value)
                or urlh.get_header('x-api-token')
                or self._search_json(
                    r'<script[^>]+\bid="shoebox-tokens"[^>]*>', webpage, 'shoebox tokens',
                    display_id, default={}).get('apiToken')
                or self._search_regex(r'\\"tokens\\":{\\"guest\\":\\"([A-Za-z0-9._-]+)\\"', webpage, 'token', default=None))

        if not token:
            self.report_warning('Unable to extract auth token')
            return None
        return {'Authorization': f'Bearer {token}'}

    def _extract_video(self, data, video_id):
        formats = []
        video_data = traverse_obj(data, ('_embedded', 'video', '_embedded', {dict}))
        m3u8_url = traverse_obj(video_data, ('playlists', 0, '_links', 'related', 'href', {url_or_none}))
        if m3u8_url:
            formats.extend(self._extract_m3u8_formats(m3u8_url, video_id, 'mp4', m3u8_id='hls', fatal=False))
        mp4_entry = traverse_obj(video_data, ('video-outputs', lambda _, v: url_or_none(v['_links']['related']['href']), any))
        if mp4_entry:
            formats.append({
                'url': mp4_entry['_links']['related']['href'],
                'format_id': 'direct',
                'quality': 1,
                **traverse_obj(mp4_entry, {
                    'height': ('height', {int_or_none}),
                    'width': ('width', {int_or_none}),
                    'filesize': ('file_size', {int_or_none}),
                }),
            })
        return {
            'id': video_id,
            'formats': formats,