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

Code before:

            pages_total = int_or_none(raw_comments.get('totalPages')) or None
            is_last = raw_comments.get('last')
            if not raw_comments.get('content') or is_last or (page_n > pages_total if pages_total else is_last is not False):
                return

    def _perform_login(self, username, password):
        LOGIN_PAGE_URL = 'https://secure.rokfin.com/auth/realms/rokfin-web/protocol/openid-connect/auth?client_id=web&redirect_uri=https%3A%2F%2Frokfin.com%2Ffeed&response_mode=fragment&response_type=code&scope=openid'
        AUTHENTICATION_URL_REGEX_STEP_1 = r'\<form\s+[^>]+action\s*=\s*"(?P<authentication_point_url>https://secure\.rokfin\.com/auth/realms/rokfin-web/login-actions/authenticate\?[^"]+)"[^>]*>'

        # In OpenID terms (https://openid.net/specs/), this program acts as a (public) Client (a.k.a. Client Application),
        # the User Agent, and the Relying Party (RP) simultaneously, with the Identity Provider (IDP) being secure.rokfin.com.
        # Rokfin uses KeyCloak (https://www.keycloak.org) as the OpenID implementation of choice.
        #
        # ---------------------------- CODE FLOW AUTHORIZATION ----------------------------
        # https://openid.net/specs/openid-connect-core-1_0.html#CodeFlowAuth (Sec. 3.1.1)
        #
        # Authentication phase:
        #
        # Step 1: preparation
        login_page = self._download_webpage(LOGIN_PAGE_URL, None, note='loading login page', fatal=False)
        if not login_page:
            return
        authentication_point_url = unescapeHTML(re.search(AUTHENTICATION_URL_REGEX_STEP_1, login_page).group('authentication_point_url'))
        if authentication_point_url is None:
            self.report_warning('login failed unexpectedly: Rokfin extractor must be updated')
            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;

Review comment (suggestion):
seems unnecessary. The next `_download_webpage` will raise appropriate error

Code after:
            if not raw_comments.get('content') or is_last or (page_n > pages_total if pages_total else is_last is not False):
                return

    def _perform_login(self, username, password):
        # https://openid.net/specs/openid-connect-core-1_0.html#CodeFlowAuth (Sec. 3.1)
        login_page = self._download_webpage(
            f'{self._AUTH_BASE}/auth?client_id=web&redirect_uri=https%3A%2F%2Frokfin.com%2Ffeed&response_mode=fragment&response_type=code&scope=openid',
            None, note='loading login page', errnote='error loading login page')
        authentication_point_url = unescapeHTML(self._search_regex(
            r'<form\s+[^>]+action\s*=\s*"(https://secure\.rokfin\.com/auth/realms/rokfin-web/login-actions/authenticate\?[^"]+)"',
            login_page, name='Authentication URL'))

        resp_body = self._download_webpage(
            authentication_point_url, None, note='logging in', fatal=False, expected_status=404,
            data=urlencode_postdata({'username': username, 'password': password, 'rememberMe': 'off', 'credentialId': ''}))
        if not self._authentication_active():
            if re.search(r'(?i)(invalid\s+username\s+or\s+password)', resp_body or ''):
                raise ExtractorError('invalid username/password', expected=True)
            raise ExtractorError('Login failed')

        urlh = self._request_webpage(
            f'{self._AUTH_BASE}/auth', None,
            note='granting user authorization', errnote='user authorization rejected by Rokfin',
            query={
                'client_id': 'web',
                'prompt': 'none',
                '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