diff --git a/yt_dlp/extractor/canalsurmas.py b/yt_dlp/extractor/canalsurmas.py index 210973a0b8..4043418524 100644 --- a/yt_dlp/extractor/canalsurmas.py +++ b/yt_dlp/extractor/canalsurmas.py @@ -1,11 +1,10 @@ import json -import time from .common import InfoExtractor from ..utils import ( determine_ext, float_or_none, - jwt_decode_hs256, + jwt_is_expired, parse_iso8601, url_or_none, variadic, @@ -31,12 +30,8 @@ class CanalsurmasIE(InfoExtractor): _API_BASE = 'https://api-rtva.interactvty.com' _access_token = None - @staticmethod - def _is_jwt_expired(token): - return jwt_decode_hs256(token)['exp'] - time.time() < 300 - def _call_api(self, endpoint, video_id, fields=None): - if not self._access_token or self._is_jwt_expired(self._access_token): + if not self._access_token or jwt_is_expired(self._access_token): self._access_token = self._download_json( f'{self._API_BASE}/jwt/token/', None, 'Downloading access token', 'Failed to download access token', diff --git a/yt_dlp/extractor/cbc.py b/yt_dlp/extractor/cbc.py index 319771655e..ac60252cc5 100644 --- a/yt_dlp/extractor/cbc.py +++ b/yt_dlp/extractor/cbc.py @@ -11,7 +11,7 @@ float_or_none, int_or_none, js_to_json, - jwt_decode_hs256, + jwt_is_expired, mimetype2ext, orderedSet, parse_age_limit, @@ -620,9 +620,6 @@ def _ropc_settings(self): 'https://services.radio-canada.ca/ott/catalog/v1/gem/settings', None, 'Downloading site settings', query={'device': 'web'})['identityManagement']['ropc'] - def _is_jwt_expired(self, token): - return jwt_decode_hs256(token)['exp'] - time.time() < 300 - def _call_oauth_api(self, oauth_data, note='Refreshing access token'): response = self._download_json( self._ropc_settings['url'], None, note, data=urlencode_postdata({ @@ -657,7 +654,7 @@ def _perform_login(self, username, password): raise def _fetch_access_token(self): - if self._is_jwt_expired(self._access_token): + if jwt_is_expired(self._access_token): try: self._call_oauth_api({ 'grant_type': 'refresh_token', @@ -675,7 +672,7 @@ def _fetch_claims_token(self): if not self._get_login_info()[0]: return None - if not self._claims_token or self._is_jwt_expired(self._claims_token): + if not self._claims_token or jwt_is_expired(self._claims_token): self._claims_token = self._download_json( 'https://services.radio-canada.ca/ott/subscription/v2/gem/Subscriber/profile', None, 'Downloading claims token', query={'device': 'web'}, diff --git a/yt_dlp/extractor/cda.py b/yt_dlp/extractor/cda.py index 027b37d448..0a98e37119 100644 --- a/yt_dlp/extractor/cda.py +++ b/yt_dlp/extractor/cda.py @@ -16,6 +16,8 @@ determine_ext, float_or_none, int_or_none, + jwt_encode_hs256, + jwt_is_expired, merge_dicts, multipart_encode, parse_duration, @@ -152,7 +154,7 @@ def _perform_login(self, username, password): self._API_HEADERS['User-Agent'] = f'pl.cda 1.0 (version {app_version}; Android {android_version}; {phone_model})' cached_bearer = self.cache.load(self._BEARER_CACHE, username) or {} - if cached_bearer.get('valid_until', 0) > dt.datetime.now().timestamp() + 5: + if not jwt_is_expired(jwt_encode_hs256(cached_bearer, 'cda.pl'), 5, 'valid_until'): self._API_HEADERS['Authorization'] = f'Bearer {cached_bearer["token"]}' return diff --git a/yt_dlp/extractor/digitalconcerthall.py b/yt_dlp/extractor/digitalconcerthall.py index 4c4fe470da..adee35c218 100644 --- a/yt_dlp/extractor/digitalconcerthall.py +++ b/yt_dlp/extractor/digitalconcerthall.py @@ -1,10 +1,9 @@ -import time - from .common import InfoExtractor from ..networking.exceptions import HTTPError from ..utils import ( ExtractorError, jwt_decode_hs256, + jwt_is_expired, parse_codecs, try_get, url_or_none, @@ -84,16 +83,14 @@ class DigitalConcertHallIE(InfoExtractor): 'User-Agent': _USER_AGENT, } _access_token = None - _access_token_expiry = 0 _refresh_token = None @property def _access_token_is_expired(self): - return self._access_token_expiry - 30 <= int(time.time()) + return jwt_is_expired(self._access_token, 30) def _set_access_token(self, value): self._access_token = value - self._access_token_expiry = traverse_obj(value, ({jwt_decode_hs256}, 'exp', {int})) or 0 def _cache_tokens(self, /): self.cache.store(self._NETRC_MACHINE, 'tokens', { diff --git a/yt_dlp/extractor/iwara.py b/yt_dlp/extractor/iwara.py index 5b5c367ad8..c6ffebf5fe 100644 --- a/yt_dlp/extractor/iwara.py +++ b/yt_dlp/extractor/iwara.py @@ -1,7 +1,6 @@ import functools import hashlib import json -import time import urllib.parse from .common import InfoExtractor @@ -9,11 +8,10 @@ ExtractorError, OnDemandPagedList, int_or_none, - jwt_decode_hs256, + jwt_is_expired, mimetype2ext, qualities, traverse_obj, - try_call, unified_timestamp, ) @@ -25,7 +23,7 @@ class IwaraBaseIE(InfoExtractor): def _is_token_expired(self, token, token_type): # User token TTL == ~3 weeks, Media token TTL == ~1 hour - if (try_call(lambda: jwt_decode_hs256(token)['exp']) or 0) <= int(time.time() - 120): + if jwt_is_expired(token, 120): self.to_screen(f'{token_type} token has expired') return True diff --git a/yt_dlp/extractor/jiocinema.py b/yt_dlp/extractor/jiocinema.py index 94c85064ef..f975e74436 100644 --- a/yt_dlp/extractor/jiocinema.py +++ b/yt_dlp/extractor/jiocinema.py @@ -4,7 +4,6 @@ import random import re import string -import time from .common import InfoExtractor from ..utils import ( @@ -12,6 +11,7 @@ float_or_none, int_or_none, jwt_decode_hs256, + jwt_is_expired, parse_age_limit, try_call, url_or_none, @@ -106,8 +106,9 @@ def _call_login_api(self, endpoint, guest_token, data, note): 'os': ('os', {str}), })}, data=data) - def _is_token_expired(self, token): - return (try_call(lambda: jwt_decode_hs256(token)['exp']) or 0) <= int(time.time() - 180) + @staticmethod + def _is_token_expired(token): + return jwt_is_expired(token, 180) def _perform_login(self, username, password): if self._ACCESS_TOKEN and not self._is_token_expired(self._ACCESS_TOKEN): diff --git a/yt_dlp/extractor/mlb.py b/yt_dlp/extractor/mlb.py index 562b93fc78..cff5d1ea31 100644 --- a/yt_dlp/extractor/mlb.py +++ b/yt_dlp/extractor/mlb.py @@ -1,6 +1,5 @@ import json import re -import time import uuid from .common import InfoExtractor @@ -10,7 +9,7 @@ determine_ext, int_or_none, join_nonempty, - jwt_decode_hs256, + jwt_is_expired, parse_duration, parse_iso8601, try_get, @@ -350,11 +349,10 @@ class MLBTVIE(InfoExtractor): _device_id = None _session_id = None _access_token = None - _token_expiry = 0 @property def _api_headers(self): - if (self._token_expiry - 120) <= time.time(): + if jwt_is_expired(self._access_token, 120): self.write_debug('Access token has expired; re-logging in') self._perform_login(*self._get_login_info()) return {'Authorization': f'Bearer {self._access_token}'} @@ -394,7 +392,6 @@ def _perform_login(self, username, password): raise ExtractorError('Invalid username or password', expected=True) raise - self._token_expiry = traverse_obj(self._access_token, ({jwt_decode_hs256}, 'exp', {int})) or 0 self._set_device_id(username) self._session_id = self._call_api({ diff --git a/yt_dlp/extractor/nfl.py b/yt_dlp/extractor/nfl.py index 59213a44be..29f5bd371c 100644 --- a/yt_dlp/extractor/nfl.py +++ b/yt_dlp/extractor/nfl.py @@ -1,7 +1,6 @@ import base64 import json import re -import time import uuid from .anvato import AnvatoIE @@ -12,6 +11,7 @@ determine_ext, get_element_by_class, int_or_none, + jwt_is_expired, make_archive_id, url_or_none, urlencode_postdata, @@ -84,7 +84,6 @@ class NFLBaseIE(InfoExtractor): _API_KEY = '3_Qa8TkWpIB8ESCBT8tY2TukbVKgO5F6BJVc7N1oComdwFzI7H2L9NOWdm11i_BY9f' _TOKEN = None - _TOKEN_EXPIRY = 0 def _get_account_info(self): cookies = self._get_cookies('https://auth-id.nfl.com/') @@ -123,7 +122,7 @@ def _get_account_info(self): raise ExtractorError('Failed to retrieve account info with provided cookies', expected=True) def _get_auth_token(self): - if self._TOKEN and self._TOKEN_EXPIRY > int(time.time() + 30): + if self._TOKEN and jwt_is_expired(self._TOKEN, 30, 'expiresIn'): return token = self._download_json( @@ -133,7 +132,6 @@ def _get_auth_token(self): data=json.dumps({**self._CLIENT_DATA, **self._ACCOUNT_INFO}, separators=(',', ':')).encode()) self._TOKEN = token['accessToken'] - self._TOKEN_EXPIRY = token['expiresIn'] self._ACCOUNT_INFO['refreshToken'] = token['refreshToken'] def _extract_video(self, mcp_id, is_live=False): diff --git a/yt_dlp/extractor/qdance.py b/yt_dlp/extractor/qdance.py index 4f71657c3f..777e2b60df 100644 --- a/yt_dlp/extractor/qdance.py +++ b/yt_dlp/extractor/qdance.py @@ -1,11 +1,10 @@ import json -import time from .common import InfoExtractor from ..utils import ( ExtractorError, int_or_none, - jwt_decode_hs256, + jwt_is_expired, str_or_none, traverse_obj, try_call, @@ -116,7 +115,7 @@ def _real_initialize(self): self.raise_login_required() def _get_auth(self): - if (try_call(lambda: jwt_decode_hs256(self._access_token)['exp']) or 0) <= int(time.time() - 120): + if jwt_is_expired(self._access_token, 120): if not self._refresh_token: raise ExtractorError( 'Cannot refresh access token, login with yt-dlp or refresh cookies in browser') diff --git a/yt_dlp/extractor/stacommu.py b/yt_dlp/extractor/stacommu.py index e6866f1517..fa778ca954 100644 --- a/yt_dlp/extractor/stacommu.py +++ b/yt_dlp/extractor/stacommu.py @@ -1,8 +1,7 @@ -import time - from .wrestleuniverse import WrestleUniverseBaseIE from ..utils import ( int_or_none, + jwt_is_expired, traverse_obj, url_basename, url_or_none, @@ -23,7 +22,7 @@ class StacommuBaseIE(WrestleUniverseBaseIE): @WrestleUniverseBaseIE._TOKEN.getter def _TOKEN(self): - if self._REAL_TOKEN and self._TOKEN_EXPIRY <= int(time.time()): + if self._REAL_TOKEN and jwt_is_expired(self._REAL_TOKEN): self._refresh_token() return self._REAL_TOKEN diff --git a/yt_dlp/extractor/vrt.py b/yt_dlp/extractor/vrt.py index 6e5514eefd..c0584cb54d 100644 --- a/yt_dlp/extractor/vrt.py +++ b/yt_dlp/extractor/vrt.py @@ -13,8 +13,8 @@ get_element_by_class, get_element_html_by_class, int_or_none, - jwt_decode_hs256, jwt_encode_hs256, + jwt_is_expired, make_archive_id, merge_dicts, parse_age_limit, @@ -304,15 +304,15 @@ def _fetch_tokens(self): access_token = self._get_vrt_cookie(self._ACCESS_TOKEN_COOKIE_NAME) video_token = self._get_vrt_cookie(self._VIDEO_TOKEN_COOKIE_NAME) - if (access_token and not self._is_jwt_token_expired(access_token) - and video_token and not self._is_jwt_token_expired(video_token)): + if (access_token and not jwt_is_expired(access_token) + and video_token and not jwt_is_expired(video_token)): return access_token, video_token if has_credentials: access_token, video_token = self.cache.load(self._NETRC_MACHINE, 'token_data', default=(None, None)) - if (access_token and not self._is_jwt_token_expired(access_token) - and video_token and not self._is_jwt_token_expired(video_token)): + if (access_token and not jwt_is_expired(access_token) + and video_token and not jwt_is_expired(video_token)): self.write_debug('Restored tokens from cache') self._set_cookie(self._TOKEN_COOKIE_DOMAIN, self._ACCESS_TOKEN_COOKIE_NAME, access_token) self._set_cookie(self._TOKEN_COOKIE_DOMAIN, self._VIDEO_TOKEN_COOKIE_NAME, video_token) @@ -347,18 +347,14 @@ def _get_vrt_cookie(self, cookie_name): # Refresh token cookie is scoped to /vrtmax/sso, others are scoped to / return try_call(lambda: self._get_cookies('https://www.vrt.be/vrtmax/sso')[cookie_name].value) - @staticmethod - def _is_jwt_token_expired(token): - return jwt_decode_hs256(token)['exp'] - time.time() < 300 - def _perform_login(self, username, password): refresh_token = self._get_vrt_cookie(self._REFRESH_TOKEN_COOKIE_NAME) - if refresh_token and not self._is_jwt_token_expired(refresh_token): + if refresh_token and not jwt_is_expired(refresh_token): self.write_debug('Using refresh token from logged-in cookies; skipping login with credentials') return refresh_token = self.cache.load(self._NETRC_MACHINE, 'refresh_token', default=None) - if refresh_token and not self._is_jwt_token_expired(refresh_token): + if refresh_token and not jwt_is_expired(refresh_token): self.write_debug('Restored refresh token from cache') self._set_cookie(self._TOKEN_COOKIE_DOMAIN, self._REFRESH_TOKEN_COOKIE_NAME, refresh_token, path='/vrtmax/sso') return diff --git a/yt_dlp/extractor/wrestleuniverse.py b/yt_dlp/extractor/wrestleuniverse.py index d401d6d39d..2b18a27ccd 100644 --- a/yt_dlp/extractor/wrestleuniverse.py +++ b/yt_dlp/extractor/wrestleuniverse.py @@ -1,7 +1,6 @@ import base64 import binascii import json -import time import uuid from .common import InfoExtractor @@ -9,7 +8,7 @@ from ..utils import ( ExtractorError, int_or_none, - jwt_decode_hs256, + jwt_is_expired, traverse_obj, try_call, url_basename, @@ -25,7 +24,6 @@ class WrestleUniverseBaseIE(InfoExtractor): _API_HOST = 'api.wrestle-universe.com' _API_PATH = None _REAL_TOKEN = None - _TOKEN_EXPIRY = None _REFRESH_TOKEN = None _DEVICE_ID = None _LOGIN_QUERY = {'key': 'AIzaSyCaRPBsDQYVDUWWBXjsTrHESi2r_F3RAdA'} @@ -40,13 +38,13 @@ class WrestleUniverseBaseIE(InfoExtractor): @property def _TOKEN(self): - if not self._REAL_TOKEN or not self._TOKEN_EXPIRY: + if not self._REAL_TOKEN: token = try_call(lambda: self._get_cookies('https://www.wrestle-universe.com/')['token'].value) if not token and not self._REFRESH_TOKEN: self.raise_login_required() self._TOKEN = token - if not self._REAL_TOKEN or self._TOKEN_EXPIRY <= int(time.time()): + if not self._REAL_TOKEN or jwt_is_expired(self._REAL_TOKEN): if not self._REFRESH_TOKEN: raise ExtractorError( 'Expired token. Refresh your cookies in browser and try again', expected=True) @@ -58,11 +56,6 @@ def _TOKEN(self): def _TOKEN(self, value): self._REAL_TOKEN = value - expiry = traverse_obj(value, ({jwt_decode_hs256}, 'exp', {int_or_none})) - if not expiry: - raise ExtractorError('There was a problem with the auth token') - self._TOKEN_EXPIRY = expiry - def _perform_login(self, username, password): login = self._download_json( 'https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword', None, diff --git a/yt_dlp/extractor/zee5.py b/yt_dlp/extractor/zee5.py index fb523de03b..1657a563e6 100644 --- a/yt_dlp/extractor/zee5.py +++ b/yt_dlp/extractor/zee5.py @@ -1,5 +1,4 @@ import json -import time import uuid from .common import InfoExtractor @@ -7,6 +6,7 @@ ExtractorError, int_or_none, jwt_decode_hs256, + jwt_is_expired, parse_age_limit, str_or_none, try_call, @@ -124,10 +124,9 @@ def _perform_login(self, username, password): else: raise ExtractorError(self._LOGIN_HINT, expected=True) - token = jwt_decode_hs256(self._USER_TOKEN) - if token.get('exp', 0) <= int(time.time()): + if jwt_is_expired(self._USER_TOKEN): raise ExtractorError('User token has expired', expected=True) - self._USER_COUNTRY = token.get('current_country') + self._USER_COUNTRY = jwt_decode_hs256(self._USER_TOKEN).get('current_country') def _real_extract(self, url): video_id, display_id = self._match_valid_url(url).group('id', 'display_id') diff --git a/yt_dlp/utils/_utils.py b/yt_dlp/utils/_utils.py index 20aa341ca3..4a65a94e7d 100644 --- a/yt_dlp/utils/_utils.py +++ b/yt_dlp/utils/_utils.py @@ -4764,6 +4764,11 @@ def jwt_decode_hs256(jwt): return json.loads(base64.urlsafe_b64decode(f'{payload_b64}===')) +def jwt_is_expired(token, buffer=300, key='exp'): + exp = traversal.traverse_obj(token, ({jwt_decode_hs256}, key, {int, float})) or 0 + return exp - time.time() < buffer + + WINDOWS_VT_MODE = False if os.name == 'nt' else None