yt_dlp/extractor/goplay.py

   1 import base64
   2 import binascii
   3 import datetime
   4 import hashlib
   5 import hmac
   6 import json
   7 import os
   8
   9 from .common import InfoExtractor
  10 from ..utils import (
  11     ExtractorError,
  12     traverse_obj,
  13     unescapeHTML,
  14 )
  15
  16
  17 class GoPlayIE(InfoExtractor):
  18     _VALID_URL = r'https?://(www\.)?goplay\.be/video/([^/]+/[^/]+/|)(?P<display_id>[^/#]+)'
  19
  20     _NETRC_MACHINE = 'goplay'
  21
  22     _TESTS = [{
  23         'url': 'https://www.goplay.be/video/de-container-cup/de-container-cup-s3/de-container-cup-s3-aflevering-2#autoplay',
  24         'info_dict': {
  25             'id': '9c4214b8-e55d-4e4b-a446-f015f6c6f811',
  26             'ext': 'mp4',
  27             'title': 'S3 - Aflevering 2',
  28             'series': 'De Container Cup',
  29             'season': 'Season 3',
  30             'season_number': 3,
  31             'episode': 'Episode 2',
  32             'episode_number': 2,
  33         },
  34         'skip': 'This video is only available for registered users'
  35     }, {
  36         'url': 'https://www.goplay.be/video/a-family-for-thr-holidays-s1-aflevering-1#autoplay',
  37         'info_dict': {
  38             'id': '74e3ed07-748c-49e4-85a0-393a93337dbf',
  39             'ext': 'mp4',
  40             'title': 'A Family for the Holidays',
  41         },
  42         'skip': 'This video is only available for registered users'
  43     }]
  44
  45     _id_token = None
  46
  47     def _perform_login(self, username, password):
  48         self.report_login()
  49         aws = AwsIdp(ie=self, pool_id='eu-west-1_dViSsKM5Y', client_id='6s1h851s8uplco5h6mqh1jac8m')
  50         self._id_token, _ = aws.authenticate(username=username, password=password)
  51
  52     def _real_initialize(self):
  53         if not self._id_token:
  54             raise self.raise_login_required(method='password')
  55
  56     def _real_extract(self, url):
  57         url, display_id = self._match_valid_url(url).group(0, 'display_id')
  58         webpage = self._download_webpage(url, display_id)
  59         video_data_json = self._html_search_regex(r'<div\s+data-hero="([^"]+)"', webpage, 'video_data')
  60         video_data = self._parse_json(unescapeHTML(video_data_json), display_id).get('data')
  61
  62         movie = video_data.get('movie')
  63         if movie:
  64             video_id = movie['videoUuid']
  65             info_dict = {
  66                 'title': movie.get('title')
  67             }
  68         else:
  69             episode = traverse_obj(video_data, ('playlists', ..., 'episodes', lambda _, v: v['pageInfo']['url'] == url), get_all=False)
  70             video_id = episode['videoUuid']
  71             info_dict = {
  72                 'title': episode.get('episodeTitle'),
  73                 'series': traverse_obj(episode, ('program', 'title')),
  74                 'season_number': episode.get('seasonNumber'),
  75                 'episode_number': episode.get('episodeNumber'),
  76             }
  77
  78         api = self._download_json(
  79             f'https://api.goplay.be/web/v1/videos/long-form/{video_id}',
  80             video_id, headers={'Authorization': 'Bearer %s' % self._id_token})
  81
  82         formats, subs = self._extract_m3u8_formats_and_subtitles(
  83             api['manifestUrls']['hls'], video_id, ext='mp4', m3u8_id='HLS')
  84
  85         info_dict.update({
  86             'id': video_id,
  87             'formats': formats,
  88         })
  89
  90         return info_dict
  91
  92
  93 # Taken from https://github.com/add-ons/plugin.video.viervijfzes/blob/master/resources/lib/viervijfzes/auth_awsidp.py
  94 # Released into Public domain by https://github.com/michaelarnauts
  95
  96 class InvalidLoginException(ExtractorError):
  97     """ The login credentials are invalid """
  98
  99
 100 class AuthenticationException(ExtractorError):
 101     """ Something went wrong while logging in """
 102
 103
 104 class AwsIdp:
 105     """ AWS Identity Provider """
 106
 107     def __init__(self, ie, pool_id, client_id):
 108         """
 109         :param InfoExtrator ie: The extractor that instantiated this class.
 110         :param str pool_id:     The AWS user pool to connect to (format: <region>_<poolid>).
 111                                 E.g.: eu-west-1_aLkOfYN3T
 112         :param str client_id:   The client application ID (the ID of the application connecting)
 113         """
 114
 115         self.ie = ie
 116
 117         self.pool_id = pool_id
 118         if "_" not in self.pool_id:
 119             raise ValueError("Invalid pool_id format. Should be <region>_<poolid>.")
 120
 121         self.client_id = client_id
 122         self.region = self.pool_id.split("_")[0]
 123         self.url = "https://cognito-idp.%s.amazonaws.com/" % (self.region,)
 124
 125         # Initialize the values
 126         # https://github.com/aws/amazon-cognito-identity-js/blob/master/src/AuthenticationHelper.js#L22
 127         self.n_hex = 'FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1' + \
 128                      '29024E088A67CC74020BBEA63B139B22514A08798E3404DD' + \
 129                      'EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245' + \
 130                      'E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED' + \
 131                      'EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D' + \
 132                      'C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F' + \
 133                      '83655D23DCA3AD961C62F356208552BB9ED529077096966D' + \
 134                      '670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B' + \
 135                      'E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9' + \
 136                      'DE2BCBF6955817183995497CEA956AE515D2261898FA0510' + \
 137                      '15728E5A8AAAC42DAD33170D04507A33A85521ABDF1CBA64' + \
 138                      'ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7' + \
 139                      'ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6B' + \
 140                      'F12FFA06D98A0864D87602733EC86A64521F2B18177B200C' + \
 141                      'BBE117577A615D6C770988C0BAD946E208E24FA074E5AB31' + \
 142                      '43DB5BFCE0FD108E4B82D120A93AD2CAFFFFFFFFFFFFFFFF'
 143
 144         # https://github.com/aws/amazon-cognito-identity-js/blob/master/src/AuthenticationHelper.js#L49
 145         self.g_hex = '2'
 146         self.info_bits = bytearray('Caldera Derived Key', 'utf-8')
 147
 148         self.big_n = self.__hex_to_long(self.n_hex)
 149         self.g = self.__hex_to_long(self.g_hex)
 150         self.k = self.__hex_to_long(self.__hex_hash('00' + self.n_hex + '0' + self.g_hex))
 151         self.small_a_value = self.__generate_random_small_a()
 152         self.large_a_value = self.__calculate_a()
 153
 154     def authenticate(self, username, password):
 155         """ Authenticate with a username and password. """
 156         # Step 1: First initiate an authentication request
 157         auth_data_dict = self.__get_authentication_request(username)
 158         auth_data = json.dumps(auth_data_dict).encode("utf-8")
 159         auth_headers = {
 160             "X-Amz-Target": "AWSCognitoIdentityProviderService.InitiateAuth",
 161             "Accept-Encoding": "identity",
 162             "Content-Type": "application/x-amz-json-1.1"
 163         }
 164         auth_response_json = self.ie._download_json(
 165             self.url, None, data=auth_data, headers=auth_headers,
 166             note='Authenticating username', errnote='Invalid username')
 167         challenge_parameters = auth_response_json.get("ChallengeParameters")
 168
 169         if auth_response_json.get("ChallengeName") != "PASSWORD_VERIFIER":
 170             raise AuthenticationException(auth_response_json["message"])
 171
 172         # Step 2: Respond to the Challenge with a valid ChallengeResponse
 173         challenge_request = self.__get_challenge_response_request(challenge_parameters, password)
 174         challenge_data = json.dumps(challenge_request).encode("utf-8")
 175         challenge_headers = {
 176             "X-Amz-Target": "AWSCognitoIdentityProviderService.RespondToAuthChallenge",
 177             "Content-Type": "application/x-amz-json-1.1"
 178         }
 179         auth_response_json = self.ie._download_json(
 180             self.url, None, data=challenge_data, headers=challenge_headers,
 181             note='Authenticating password', errnote='Invalid password')
 182
 183         if 'message' in auth_response_json:
 184             raise InvalidLoginException(auth_response_json['message'])
 185         return (
 186             auth_response_json['AuthenticationResult']['IdToken'],
 187             auth_response_json['AuthenticationResult']['RefreshToken']
 188         )
 189
 190     def __get_authentication_request(self, username):
 191         """
 192
 193         :param str username:    The username to use
 194
 195         :return: A full Authorization request.
 196         :rtype: dict
 197         """
 198         auth_request = {
 199             "AuthParameters": {
 200                 "USERNAME": username,
 201                 "SRP_A": self.__long_to_hex(self.large_a_value)
 202             },
 203             "AuthFlow": "USER_SRP_AUTH",
 204             "ClientId": self.client_id
 205         }
 206         return auth_request
 207
 208     def __get_challenge_response_request(self, challenge_parameters, password):
 209         """ Create a Challenge Response Request object.
 210
 211         :param dict[str,str|imt] challenge_parameters:  The parameters for the challenge.
 212         :param str password:                            The password.
 213
 214         :return: A valid and full request data object to use as a response for a challenge.
 215         :rtype: dict
 216         """
 217         user_id = challenge_parameters["USERNAME"]
 218         user_id_for_srp = challenge_parameters["USER_ID_FOR_SRP"]
 219         srp_b = challenge_parameters["SRP_B"]
 220         salt = challenge_parameters["SALT"]
 221         secret_block = challenge_parameters["SECRET_BLOCK"]
 222
 223         timestamp = self.__get_current_timestamp()
 224
 225         # Get a HKDF key for the password, SrpB and the Salt
 226         hkdf = self.__get_hkdf_key_for_password(
 227             user_id_for_srp,
 228             password,
 229             self.__hex_to_long(srp_b),
 230             salt
 231         )
 232         secret_block_bytes = base64.standard_b64decode(secret_block)
 233
 234         # the message is a combo of the pool_id, provided SRP userId, the Secret and Timestamp
 235         msg = \
 236             bytearray(self.pool_id.split('_')[1], 'utf-8') + \
 237             bytearray(user_id_for_srp, 'utf-8') + \
 238             bytearray(secret_block_bytes) + \
 239             bytearray(timestamp, 'utf-8')
 240         hmac_obj = hmac.new(hkdf, msg, digestmod=hashlib.sha256)
 241         signature_string = base64.standard_b64encode(hmac_obj.digest()).decode('utf-8')
 242         challenge_request = {
 243             "ChallengeResponses": {
 244                 "USERNAME": user_id,
 245                 "TIMESTAMP": timestamp,
 246                 "PASSWORD_CLAIM_SECRET_BLOCK": secret_block,
 247                 "PASSWORD_CLAIM_SIGNATURE": signature_string
 248             },
 249             "ChallengeName": "PASSWORD_VERIFIER",
 250             "ClientId": self.client_id
 251         }
 252         return challenge_request
 253
 254     def __get_hkdf_key_for_password(self, username, password, server_b_value, salt):
 255         """ Calculates the final hkdf based on computed S value, and computed U value and the key.
 256
 257         :param str username:        Username.
 258         :param str password:        Password.
 259         :param int server_b_value:  Server B value.
 260         :param int salt:            Generated salt.
 261
 262         :return Computed HKDF value.
 263         :rtype: object
 264         """
 265
 266         u_value = self.__calculate_u(self.large_a_value, server_b_value)
 267         if u_value == 0:
 268             raise ValueError('U cannot be zero.')
 269         username_password = '%s%s:%s' % (self.pool_id.split('_')[1], username, password)
 270         username_password_hash = self.__hash_sha256(username_password.encode('utf-8'))
 271
 272         x_value = self.__hex_to_long(self.__hex_hash(self.__pad_hex(salt) + username_password_hash))
 273         g_mod_pow_xn = pow(self.g, x_value, self.big_n)
 274         int_value2 = server_b_value - self.k * g_mod_pow_xn
 275         s_value = pow(int_value2, self.small_a_value + u_value * x_value, self.big_n)
 276         hkdf = self.__compute_hkdf(
 277             bytearray.fromhex(self.__pad_hex(s_value)),
 278             bytearray.fromhex(self.__pad_hex(self.__long_to_hex(u_value)))
 279         )
 280         return hkdf
 281
 282     def __compute_hkdf(self, ikm, salt):
 283         """ Standard hkdf algorithm
 284
 285         :param {Buffer} ikm Input key material.
 286         :param {Buffer} salt Salt value.
 287         :return {Buffer} Strong key material.
 288         """
 289
 290         prk = hmac.new(salt, ikm, hashlib.sha256).digest()
 291         info_bits_update = self.info_bits + bytearray(chr(1), 'utf-8')
 292         hmac_hash = hmac.new(prk, info_bits_update, hashlib.sha256).digest()
 293         return hmac_hash[:16]
 294
 295     def __calculate_u(self, big_a, big_b):
 296         """ Calculate the client's value U which is the hash of A and B
 297
 298         :param int big_a:   Large A value.
 299         :param int big_b:   Server B value.
 300
 301         :return Computed U value.
 302         :rtype: int
 303         """
 304
 305         u_hex_hash = self.__hex_hash(self.__pad_hex(big_a) + self.__pad_hex(big_b))
 306         return self.__hex_to_long(u_hex_hash)
 307
 308     def __generate_random_small_a(self):
 309         """ Helper function to generate a random big integer
 310
 311         :return a random value.
 312         :rtype: int
 313         """
 314         random_long_int = self.__get_random(128)
 315         return random_long_int % self.big_n
 316
 317     def __calculate_a(self):
 318         """ Calculate the client's public value A = g^a%N with the generated random number a
 319
 320         :return Computed large A.
 321         :rtype: int
 322         """
 323
 324         big_a = pow(self.g, self.small_a_value, self.big_n)
 325         # safety check
 326         if (big_a % self.big_n) == 0:
 327             raise ValueError('Safety check for A failed')
 328         return big_a
 329
 330     @staticmethod
 331     def __long_to_hex(long_num):
 332         return '%x' % long_num
 333
 334     @staticmethod
 335     def __hex_to_long(hex_string):
 336         return int(hex_string, 16)
 337
 338     @staticmethod
 339     def __hex_hash(hex_string):
 340         return AwsIdp.__hash_sha256(bytearray.fromhex(hex_string))
 341
 342     @staticmethod
 343     def __hash_sha256(buf):
 344         """AuthenticationHelper.hash"""
 345         digest = hashlib.sha256(buf).hexdigest()
 346         return (64 - len(digest)) * '0' + digest
 347
 348     @staticmethod
 349     def __pad_hex(long_int):
 350         """ Converts a Long integer (or hex string) to hex format padded with zeroes for hashing
 351
 352         :param int|str long_int:    Number or string to pad.
 353
 354         :return Padded hex string.
 355         :rtype: str
 356         """
 357
 358         if not isinstance(long_int, str):
 359             hash_str = AwsIdp.__long_to_hex(long_int)
 360         else:
 361             hash_str = long_int
 362         if len(hash_str) % 2 == 1:
 363             hash_str = '0%s' % hash_str
 364         elif hash_str[0] in '89ABCDEFabcdef':
 365             hash_str = '00%s' % hash_str
 366         return hash_str
 367
 368     @staticmethod
 369     def __get_random(nbytes):
 370         random_hex = binascii.hexlify(os.urandom(nbytes))
 371         return AwsIdp.__hex_to_long(random_hex)
 372
 373     @staticmethod
 374     def __get_current_timestamp():
 375         """ Creates a timestamp with the correct English format.
 376
 377         :return: timestamp in format 'Sun Jan 27 19:00:04 UTC 2019'
 378         :rtype: str
 379         """
 380
 381         # We need US only data, so we cannot just do a strftime:
 382         # Sun Jan 27 19:00:04 UTC 2019
 383         months = [None, 'Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
 384         days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
 385
 386         time_now = datetime.datetime.now(datetime.timezone.utc)
 387         format_string = "{} {} {} %H:%M:%S UTC %Y".format(days[time_now.weekday()], months[time_now.month], time_now.day)
 388         time_string = time_now.strftime(format_string)
 389         return time_string
 390
 391     def __str__(self):
 392         return "AWS IDP Client for:\nRegion: %s\nPoolId: %s\nAppId:  %s" % (
 393             self.region, self.pool_id.split("_")[1], self.client_id
 394         )