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

Code before:
            'categories': gif_data.get('tags') or [],
            'tags': gif_data.get('tags'),
            'age_limit': 18,
            'formats': formats,
        }

    def _fetch_oauth_token(self, video_id):
        # https://github.com/Redgifs/api/wiki/Temporary-tokens
        auth = self._download_json('https://api.redgifs.com/v2/auth/temporary',
                                   video_id, note='Fetching temporary token')
        if not auth.get('token'):
            raise ExtractorError('Unable to get temporary token')
        self._API_HEADERS['authorization'] = f'Bearer {auth["token"]}'

    def _call_api(self, ep, video_id, *args, **kwargs):
        if 'authorization' not in self._API_HEADERS:
            self._fetch_oauth_token(video_id)
        assert 'authorization' in self._API_HEADERS

        for _ in range(MAX_TOKEN_REFRESH_ATTEMPTS + 1):
            try:
                headers = dict(self._API_HEADERS)
                headers['x-customheader'] = f'https://www.redgifs.com/watch/{video_id}'

                data = self._download_json(
                    f'https://api.redgifs.com/v2/{ep}', video_id, headers=headers, *args, **kwargs)
                break
            except ExtractorError as e:
                if not isinstance(e.cause, urllib.error.HTTPError) or e.cause.code != 401:
                    raise

                # Likely the temporary token has expired
                del self._API_HEADERS['authorization']
                self._fetch_oauth_token(video_id)

        if 'error' in data:
            raise ExtractorError(f'RedGifs said: {data["error"]}', expected=True, video_id=video_id)
        return data

    def _fetch_page(self, ep, video_id, query, page):
        query['page'] = page + 1
        data = self._call_api(
            ep, video_id, query=query, note=f'Downloading JSON metadata page {page + 1}')

        for entry in data['gifs']:
            yield self._parse_gif_data(entry)

    def _prepare_api_query(self, query, fields):
        api_query = [
            (field_name, query.get(field_name, (default,))[0])
            for field_name, default in fields.items()]

Review comment (security):
 - the first `_fetch_oauth_token` call can be moved inside the loop to get rid of the duplication
 - the `assert` line can also be removed since `_fetch_oauth_token` is fatal
 - `data` should be defined before `try/except` in case there are two consecutive 401 responses for some unforeseen reason
```py
    def _call_api(self, ep, video_id, *args, **kwargs):
        data = {'error': '401 Unauthorized'}  # in case of two 401 responses
        for _ in range(2):
            if 'authorization' not in self._API_HEADERS:
                self._fetch_oauth_token(video_id)
            try:
                headers = dict(self._API_HEADERS)
                headers['x-customheader'] = f'https://www.redgifs.com/watch/{video_id}'
                data = self._download_json(
                    f'https://api.redgifs.com/v2/{ep}', video_id, headers=headers, *args, **kwargs)
                break
            except ExtractorError as e:
                if not isinstance(e.cause, urllib.error.HTTPError) or e.cause.code != 401:
                    raise
                del self._API_HEADERS['authorization']  # refresh the token

        if 'error' in data:
            raise ExtractorError(f'RedGifs said: {data["error"]}', video_id=video_id)
        return data
```



Code after:
            'formats': formats,
        }

    def _fetch_oauth_token(self, video_id):
        # https://github.com/Redgifs/api/wiki/Temporary-tokens
        auth = self._download_json('https://api.redgifs.com/v2/auth/temporary',
                                   video_id, note='Fetching temporary token')
        if not auth.get('token'):
            raise ExtractorError('Unable to get temporary token')
        self._API_HEADERS['authorization'] = f'Bearer {auth["token"]}'

    def _call_api(self, ep, video_id, *args, **kwargs):
        for attempt in range(2):
            if 'authorization' not in self._API_HEADERS:
                self._fetch_oauth_token(video_id)
            try:
                headers = dict(self._API_HEADERS)
                headers['x-customheader'] = f'https://www.redgifs.com/watch/{video_id}'
                data = self._download_json(
                    f'https://api.redgifs.com/v2/{ep}', video_id, headers=headers, *args, **kwargs)
                break
            except ExtractorError as e:
                if not attempt and isinstance(e.cause, urllib.error.HTTPError) and e.cause.code == 401:
                    del self._API_HEADERS['authorization']  # refresh the token
                raise

        if 'error' in data:
            raise ExtractorError(f'RedGifs said: {data["error"]}', expected=True, video_id=video_id)
        return data

    def _fetch_page(self, ep, video_id, query, page):
        query['page'] = page + 1
        data = self._call_api(
            ep, video_id, query=query, note=f'Downloading JSON metadata page {page + 1}')

        for entry in data['gifs']:
            yield self._parse_gif_data(entry)

    def _prepare_api_query(self, query, fields):
        api_query = [
            (field_name, query.get(field_name, (default,))[0])
            for field_name, default in fields.items()]

        return {key: val for key, val in api_query if val is not None}

    def _paged_entries(self, ep, item_id, query, fields):
        page = int_or_none(query.get('page', (None,))[0])
        page_fetcher = functools.partial(
            self._fetch_page, ep, item_id, self._prepare_api_query(query, fields))
        return page_fetcher(page) if page else OnDemandPagedList(page_fetcher, self._PAGE_SIZE)