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

Code before:
            self._clear_cookies()
            return

        # Step 2 & 3: authentication
        resp_body = self._download_webpage(
            authentication_point_url, None, note='logging in', fatal=False, expected_status=404, encoding='utf-8',
            data=urlencode_postdata({'username': username, 'password': password, 'rememberMe': 'off', 'credentialId': ''}))
        # rememberMe=off resets the session when yt-dlp exits:
        # https://web.archive.org/web/20220218003425/https://wjw465150.gitbooks.io/keycloak-documentation/content/server_admin/topics/login-settings/remember-me.html
        if not self._authentication_active():
            self._clear_cookies()
            self.report_warning('login failed' + (': invalid username or password.' if type(resp_body) is str and re.search(r'invalid\s+username\s+or\s+password', resp_body, re.IGNORECASE) else ''))
            return

        # Authorization phase:
        #
        # Steps 4-7:
        access_mgmt_tokens = self._get_OAuth_tokens()
        if not access_mgmt_tokens:
            self._logout()
            return

        # 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

Review comment (security):
```suggestion
        access_mgmt_tokens = self._get_OAuth_tokens()
        if not access_mgmt_tokens:
            self._logout()
            return
```

Code after:
                'redirect_uri': 'https://rokfin.com/silent-check-sso.html',
                'response_mode': 'fragment',
                'response_type': 'code',
                'scope': 'openid',
            })
        self._access_mgmt_tokens = self._download_json(
            f'{self._AUTH_BASE}/token', None,
            note='getting access credentials', errnote='error getting access credentials',
            data=urlencode_postdata({
                'code': urllib.parse.parse_qs(urllib.parse.urldefrag(urlh.geturl()).fragment).get('code')[0],
                'client_id': 'web',
                'grant_type': 'authorization_code',
                'redirect_uri': 'https://rokfin.com/silent-check-sso.html'
            }))

    def _authentication_active(self):
        return not (
            {'KEYCLOAK_IDENTITY', 'KEYCLOAK_IDENTITY_LEGACY', 'KEYCLOAK_SESSION', 'KEYCLOAK_SESSION_LEGACY'}
            - set(self._get_cookies(self._AUTH_BASE)))

    def _get_auth_token(self):
        return try_get(self._access_mgmt_tokens, lambda x: ' '.join([x['token_type'], x['access_token']]))

    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)