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

Code before:

    def _real_extract(self, url):
        service, creator_slug = self._match_valid_url(url).group('service', 'id')

        encoded_creator_slug = creator_slug.replace('/', '%23')
        creator_id = self._call_api(
            f'get_user_by_social_media/{service}/{encoded_creator_slug}',
            creator_slug, note='Fetching user ID')

        data = self._call_api(
            f'get_livestream_channel_info/{creator_id}', creator_id,
            note='Fetching user profile feed')['channel_info']

        if not traverse_obj(data, ('channel', 'is_live', {bool})):
            raise UserNotLive(video_id=creator_id)

        stream_data = traverse_obj(data, ('channel', {
            'url': ('playback_url', {url_or_none}, {require('Stream url')}),
            'token': ('playback_auth_token', {str}, {require('Stream token')}),
        }))

        formats = self._extract_m3u8_formats(
            stream_data['url'], creator_slug, live=True, query={
                'token': stream_data['token'],
                'player_version': '1.17.0',
            })

        return {
            'id': creator_slug,
            'formats': formats,
            'is_live': True,
            **traverse_obj(data, {
                'title': ('livestream_event_info', 'event_name', {str}),
                'description': ('livestream_event_info', 'event_description', {str}),
                'thumbnail': ('livestream_event_info', 'livestream_preview_file', {url_or_none}),
                'timestamp': ('stream', 'start_time', {parse_iso8601}),
                'view_count': ('stream', 'viewer_count', {int_or_none}),
            }),
        }


Review comment (security):
If both the url and token are required, we don't need the traversal/arbitrary dict here, we can just let the KeyErrors rip
```suggestion
        channel_info = data['channel']
        formats = self._extract_m3u8_formats(
            channel_info['playback_url'], creator_slug, live=True, query={
                'token': channel_info['playback_auth_token'],
                'player_version': '1.17.0',
            })
```

Code after:

        data = self._call_api(
            f'get_livestream_channel_info/{creator_id}', creator_id,
            note='Fetching user profile feed')['channel_info']

        if not traverse_obj(data, ('channel', 'is_live', {bool})):
            raise UserNotLive(video_id=creator_id)

        channel_info = data['channel']

        return {
            'id': creator_slug,
            'formats': self._extract_m3u8_formats(
                channel_info['playback_url'], creator_slug, live=True, query={
                    'token': channel_info['playback_auth_token'],
                    'player_version': '1.17.0',
                }),
            'is_live': True,
            **traverse_obj(data, {
                'title': ('livestream_event_info', 'event_name', {str}),
                'description': ('livestream_event_info', 'event_description', {str}),
                'thumbnail': ('livestream_event_info', 'livestream_preview_file', {url_or_none}),
                'timestamp': ('stream', 'start_time', {parse_iso8601}),
                'view_count': ('stream', 'viewer_count', {int_or_none}),
            }),
        }