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

Code before:

        self.cache.store(self._NETRC_MACHINE, 'device_id', self._DEVICE_ID)

    def _refresh_token(self):
        refresh = self._download_json(
            'https://securetoken.googleapis.com/v1/token', None, 'Refreshing token',
            query=self._LOGIN_QUERY, data=urlencode_postdata({
                'grant_type': 'refresh_token',
                'refresh_token': self._REFRESH_TOKEN,
            }), headers={
                **self._LOGIN_HEADERS,
                'Content-Type': 'application/x-www-form-urlencoded',
            })
        if traverse_obj(refresh, ('refresh_token', {str})):
            self._REFRESH_TOKEN = refresh['refresh_token']
        token = traverse_obj(refresh, 'access_token', 'id_token', expected_type=str)
        if not token:
            raise ExtractorError('No auth token returned from refresh request')
        self._TOKEN = token

    def _call_api(self, video_id, param='', msg='API', auth=True, data=None, query={}, fatal=True):
        headers = {'CA-CID': ''}
        if data:
            headers['Content-Type'] = 'application/json;charset=utf-8'
            data = json.dumps(data, separators=(',', ':')).encode()
        if auth:
            headers['Authorization'] = f'Bearer {self._TOKEN}'
        return self._download_json(
            f'https://{self._API_HOST}/v1/{self._API_PATH}/{video_id}{param}', video_id,
            note=f'Downloading {msg} JSON', errnote=f'Failed to download {msg} JSON',
            data=data, headers=headers, query=query, fatal=fatal)

    def _call_encrypted_api(self, video_id, param='', msg='API', data={}, query={}, fatal=True):
        if not Cryptodome.RSA:
            raise ExtractorError('pycryptodomex not found. Please install', expected=True)
        private_key = Cryptodome.RSA.generate(2048)
        cipher = Cryptodome.PKCS1_OAEP.new(private_key, hashAlgo=Cryptodome.SHA1)

        def decrypt(data):
            if not data:
                return None
            try:
                return cipher.decrypt(base64.b64decode(data)).decode()
            except (ValueError, binascii.Error) as e:
                raise ExtractorError(f'Could not decrypt data: {e}')

        token = base64.b64encode(private_key.public_key().export_key('DER')).decode()
        api_json = self._call_api(video_id, param, msg, data={
            'deviceId': self._DEVICE_ID,
            'token': token,
            **data,

Review comment (security):
This was fine to do; for the best even. Sorry if I was being confusing w/my earlier comment
```suggestion
        if auth and self._TOKEN:
```

Code after:

        self.cache.store(self._NETRC_MACHINE, 'device_id', self._DEVICE_ID)

    def _refresh_token(self):
        refresh = self._download_json(
            'https://securetoken.googleapis.com/v1/token', None, 'Refreshing token',
            query=self._LOGIN_QUERY, data=urlencode_postdata({
                'grant_type': 'refresh_token',
                'refresh_token': self._REFRESH_TOKEN,
            }), headers={
                **self._LOGIN_HEADERS,
                'Content-Type': 'application/x-www-form-urlencoded',
            })
        if traverse_obj(refresh, ('refresh_token', {str})):
            self._REFRESH_TOKEN = refresh['refresh_token']
        token = traverse_obj(refresh, 'access_token', 'id_token', expected_type=str)
        if not token:
            raise ExtractorError('No auth token returned from refresh request')
        self._TOKEN = token

    def _call_api(self, video_id, param='', msg='API', auth=True, data=None, query={}, fatal=True):
        headers = {'CA-CID': ''}
        if data:
            headers['Content-Type'] = 'application/json;charset=utf-8'
            data = json.dumps(data, separators=(',', ':')).encode()
        if auth and self._TOKEN:
            headers['Authorization'] = f'Bearer {self._TOKEN}'
        return self._download_json(
            f'https://{self._API_HOST}/v1/{self._API_PATH}/{video_id}{param}', video_id,
            note=f'Downloading {msg} JSON', errnote=f'Failed to download {msg} JSON',
            data=data, headers=headers, query=query, fatal=fatal)

    def _call_encrypted_api(self, video_id, param='', msg='API', data={}, query={}, fatal=True):
        if not Cryptodome.RSA:
            raise ExtractorError('pycryptodomex not found. Please install', expected=True)
        private_key = Cryptodome.RSA.generate(2048)
        cipher = Cryptodome.PKCS1_OAEP.new(private_key, hashAlgo=Cryptodome.SHA1)

        def decrypt(data):
            if not data:
                return None
            try:
                return cipher.decrypt(base64.b64decode(data)).decode()
            except (ValueError, binascii.Error) as e:
                raise ExtractorError(f'Could not decrypt data: {e}')

        token = base64.b64encode(private_key.public_key().export_key('DER')).decode()
        api_json = self._call_api(video_id, param, msg, data={
            'deviceId': self._DEVICE_ID,
            'token': token,
            **data,