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

Code before:
        # Validation phase (step 8) skipped, i.e.
        #
        # (1) this extractor does not implement client-side ID-token validation;
        # (2) Rokfin does not supply Subject Identifier.

        self._access_mgmt_tokens = access_mgmt_tokens

    def _logout(self):
        LOGOUT_URL = 'https://secure.rokfin.com/auth/realms/rokfin-web/protocol/openid-connect/logout?redirect_uri=https%3A%2F%2Frokfin.com%2F'
        if self._get_login_info()[0] is None:  # username is None
            return
        self._download_webpage_handle(LOGOUT_URL, None, note='logging out', fatal=False, encoding='utf-8')
        if self._authentication_active():
            self.write_debug('logout failed')
        self._clear_cookies()
        self._access_mgmt_tokens = None
        # No token revocation takes place during logout, as KEYCLOAK does not -- and has no plans to -- support individual token
        # revocation on external party's request. See
        # https://web.archive.org/web/20220215040021/https://keycloak.discourse.group/t/revoking-or-invalidating-an-authorization-token/1032

    def _authentication_active(self):
        current_time_utc = datetime.utcnow().timestamp()
        SESSION_COOKIE_NAMES = {'KEYCLOAK_IDENTITY', 'KEYCLOAK_IDENTITY_LEGACY', 'KEYCLOAK_SESSION', 'KEYCLOAK_SESSION_LEGACY'}
        return set([cookie.name for cookie in self._downloader.cookiejar if (cookie.name in SESSION_COOKIE_NAMES)
                    and (cookie.name not in ('KEYCLOAK_SESSION', 'KEYCLOAK_SESSION_LEGACY')
                         or cookie.expires > current_time_utc)]) == SESSION_COOKIE_NAMES

    def _download_json_using_access_token(self, url_or_request, video_id, headers={}, query={}):
        assert 'authorization' not in headers
        headers = headers.copy()
        authorization_hdr_val = try_get(self._access_mgmt_tokens, lambda tokens: tokens['token_type'] + ' ' + tokens['access_token'])
        if authorization_hdr_val:
            headers['authorization'] = authorization_hdr_val
        json_string, urlh = self._download_webpage_handle(
            url_or_request, video_id, note='Downloading JSON metadata' + (' [logged in]' if 'authorization' in headers else ''),
            errnote='Unable to download JSON metadata' + (' [logged in]' if 'authorization' in headers else ''),
            headers=headers, query=query, expected_status=401)  # 401=Unauthorized
        if not authorization_hdr_val or urlh.code != 401 or self._access_mgmt_tokens.get('refresh_token') is None:
            return self._parse_json(json_string, video_id)
        del headers['authorization']
        del self._access_mgmt_tokens['access_token']
        del self._access_mgmt_tokens['token_type']
        try:
            del self._access_mgmt_tokens['expires_in']
        except KeyError:
            pass
        self._access_mgmt_tokens.update(
            self._download_json(
                self._TOKEN_DISTRIBUTION_POINT_URL_STEP_6_7, video_id, note='Restoring lost authorization',
                errnote='Failed to restore authorization', fatal=False,
                data=urlencode_postdata({'grant_type': 'refresh_token', 'refresh_token': self._access_mgmt_tokens.get('refresh_token'), 'client_id': 'web'})) or {})

Review comment (suggestion):
I think this is what u want:
```suggestion
        return bool(traverse_obj(
            self._get_cookies('https://www.rokfin.com'),
            'KEYCLOAK_IDENTITY', 'KEYCLOAK_IDENTITY_LEGACY', 'KEYCLOAK_SESSION', 'KEYCLOAK_SESSION_LEGACY'))
```

Code after:

    def _download_json_using_access_token(self, url_or_request, video_id, headers={}, query={}):
        assert 'authorization' not in headers
        headers = headers.copy()
        auth_token = self._get_auth_token()
        refresh_token = self._access_mgmt_tokens.get('refresh_token')
        if auth_token:
            headers['authorization'] = auth_token

        json_string, urlh = self._download_webpage_handle(
            url_or_request, video_id, headers=headers, query=query, expected_status=401)
        if not auth_token or urlh.code != 401 or refresh_token is None:
            return self._parse_json(json_string, video_id)

        self._access_mgmt_tokens = self._download_json(
            f'{self._AUTH_BASE}/token', video_id,
            note='User authorization expired or canceled by Rokfin. Re-authorizing ...', errnote='Failed to re-authorize',
            data=urlencode_postdata({
                'grant_type': 'refresh_token',
                'refresh_token': refresh_token,
                'client_id': 'web'
            }))
        headers['authorization'] = self._get_auth_token()
        if headers['authorization'] is None:
            raise ExtractorError('User authorization lost', expected=True)

        return self._download_json(url_or_request, video_id, headers=headers, query=query)


class RokfinPlaylistBaseIE(InfoExtractor):
    _TYPES = {
        'video': 'post',
        'audio': 'post',
        'stream': 'stream',
        'dead_stream': 'stream',
        'stack': 'stack',
    }

    def _get_video_data(self, metadata):
        for content in metadata.get('content') or []:
            media_type = self._TYPES.get(content.get('mediaType'))
            video_id = content.get('id') if media_type == 'post' else content.get('mediaId')
            if not media_type or not video_id:
                continue

            yield self.url_result(f'https://rokfin.com/{media_type}/{video_id}', video_id=f'{media_type}/{video_id}',
                                  video_title=str_or_none(traverse_obj(content, ('content', 'contentTitle'))))


class RokfinStackIE(RokfinPlaylistBaseIE):
    IE_NAME = 'rokfin:stack'