main
widevinedump 2021-12-23 15:51:08 +05:30
parent aba5a2d349
commit f910bf33ca
83 changed files with 8442 additions and 0 deletions

View File

@ -0,0 +1,8 @@
@@ECHO OFF
set/p URL="URL: "
movie.py --url %URL% -a
pause
@@ECHO OFF

8
GPLAY.bat 100644
View File

@ -0,0 +1,8 @@
@@ECHO OFF
set/p URL="URL: "
movie.py --url %URL%
pause
@@ECHO OFF

4
auth.json 100644
View File

@ -0,0 +1,4 @@
{
"authorization": "ya29.a0ARrdaM9BV33Jclsg1u_i-XYtanOmZ9cY4XYGJgkmzIq1CptxMIl0VS9I25lIYjPoJTs2EjtKVTZo8AHnqQdT7uRoZNiQ3XNyNdLQrPWjMcWmjnGgmFUzbgz0y1oB6ZukYEjayFTJt0x3euOmrXnGdHr-TPJQ",
"x-client-data": "CLC1yQEIkbbJAQiltskBCMS2yQEIqZ3KAQj4x8oBCIrbygEI0JrLAQioncsBCKCgywEIv6DLAQjc8ssBCKjzywE="
}

BIN
binaries/aria2c.exe 100644

Binary file not shown.

BIN
binaries/ffmpeg.exe 100644

Binary file not shown.

Binary file not shown.

Binary file not shown.

257
movie.py 100644
View File

@ -0,0 +1,257 @@
import argparse
import base64
import binascii
import json
import os
import requests
import subprocess
import sys
from colorama import init, Fore
from prettytable import PrettyTable
from pywidevine.decrypt.wvdecrypt import WvDecrypt
init(autoreset=True)
class Main(object):
def __init__(self, folders, args):
self.folders = folders
self.args = args
self.auth_json = None
self.movie_id = args.url.split('id=')[-1]
self.movie_details = None
self.movie_resources = {}
self.mpd_representations = {'video': [], 'audio': [], 'subtitle': []}
self.license = None
def auth(self):
if os.path.exists('auth.json'):
with open('auth.json', 'r') as src:
self.auth_json = json.loads(src.read())
else:
sys.exit()
def requests_headers(self):
return {
'accept': '*/*',
'accept-language': 'en-US,en;q=0.9',
'authorization': 'Bearer {0}'.format(self.auth_json['authorization']),
'Host': 'www.googleapis.com',
'origin': 'chrome-extension://gdijeikdkaembjbdobgfkoidjkpbmlkd',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'cross-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.141 Safari/537.36',
'x-client-data': '{0}'.format(self.auth_json['x-client-data']),
}
def get_movie_details(self):
url = 'https://www.googleapis.com/android_video/v1/asset/list?id=yt%3Amovie%3A{0}&if=imbrg&lr=en_US&cr=US&alt=json&access_token={1}&make=Google&model=ChromeCDM-Windows-x86-32&product=generic&device=generic'.format(self.movie_id, self.auth_json['authorization'])
self.movie_details = requests.get(url=url, headers=self.requests_headers()).json()
def get_movie_resources(self):
url = 'https://www.googleapis.com/android_video/v1/mpd?id=yt%3Amovie%3A{0}&ac3=true&all51=true&nd=false&all=false&secure=true&msu=false&ma=true&fc=true&hdcp=true&alt={1}&ssrc=googlevideo&access_token={2}&make=Google&model=ChromeCDM-Windows-x86-32&product=generic&device=generic'
self.movie_resources['json'] = requests.get(
url = url.format(self.movie_id, 'json', self.auth_json['authorization']),
headers = self.requests_headers()
).json()
#self.movie_resources['protojson'] = requests.get(
#url = url.format(self.movie_id, 'protojson', self.auth_json['authorization']),
#headers = self.requests_headers()
#).json()
def parse_movie_resources(self):
av_representations = self.movie_resources['json']['representations']
for x in av_representations:
if 'audio_info' not in x:
self.mpd_representations['video'].append({
'playback_url': x['playback_url'],
'codec': x['codec'],
'init': x['init'],
'bitrate': x['bitrate'],
'quality': str(x['height'])+'p',
'fps': x['video_fps']
})
elif 'audio_info' in x:
self.mpd_representations['audio'].append({
'playback_url': x['playback_url'],
'codec': x['codec'],
'init': x['init'],
'bitrate': x['bitrate'],
'language': x['audio_info']['language']
})
#subtitle_representations = self.movie_resources['protojson']['1007']['4']
#for x in subtitle_representations:
#self.mpd_representations['subtitle'].append({
#'language': x['1'],
#'url': x['3'] ,
#'format': x['5']
#})
def aria2c(self, url, output_file_name):
aria2c = os.path.join(self.folders['binaries'], 'aria2c.exe')
aria2c_command = [
aria2c, url,
'-d', self.folders['temp'], '-j16',
'-o', output_file_name, '-s16', '-x16',
'-U', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.141 Safari/537.36',
'--allow-overwrite=false',
'--async-dns=false',
'--auto-file-renaming=false',
'--console-log-level=warn',
'--retry-wait=5',
'--summary-interval=0',
]
subprocess.run(aria2c_command)
return os.path.join(self.folders['temp'], output_file_name)
def extract_pssh(self, mp4_file):
mp4dump = os.path.join(self.folders['binaries'], 'mp4dump.exe')
wv_system_id = '[ed ef 8b a9 79 d6 4a ce a3 c8 27 dc d5 1d 21 ed]'
pssh = None
data = subprocess.check_output([mp4dump, '--format', 'json', '--verbosity', '1', mp4_file])
data = json.loads(data)
for atom in data:
if atom['name'] == 'moov':
for child in atom['children']:
if child['name'] == 'pssh' and child['system_id'] == wv_system_id:
pssh = child['data'][1:-1].replace(' ', '')
pssh = binascii.unhexlify(pssh)
pssh = pssh[0:]
pssh = base64.b64encode(pssh).decode('utf-8')
return pssh
def license_request(self, pssh):
license_url = 'https://play.google.com/video/license/GetCencLicense?source=YOUTUBE&video_id={0}&oauth={1}'.format(self.movie_id, self.auth_json['authorization'])
license_headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.141 Safari/537.36'
}
wvdecrypt = WvDecrypt(pssh)
challenge = wvdecrypt.get_challenge()
resp = requests.post(url=license_url, headers=license_headers, data=challenge)
resp1 = resp.content.split('\r\n\r\n'.encode('utf-8'))
resp2 = resp1[1]
license_b64 = base64.b64encode(resp2).decode('utf-8')
wvdecrypt.update_license(license_b64)
keys = wvdecrypt.start_process()
return keys
def mp4decrypt(self, keys):
mp4decrypt_command = [os.path.join(self.folders['binaries'], 'mp4decrypt.exe')]
for key in keys:
if key.type == 'CONTENT':
mp4decrypt_command.append('--show-progress')
mp4decrypt_command.append('--key')
mp4decrypt_command.append('{}:{}'.format(key.kid.hex(), key.key.hex()))
return mp4decrypt_command
def decrypt(self, keys, input, output):
mp4decrypt_command = self.mp4decrypt(keys)
mp4decrypt_command.append(input)
mp4decrypt_command.append(output)
wvdecrypt_process = subprocess.Popen(mp4decrypt_command)
wvdecrypt_process.communicate()
wvdecrypt_process.wait()
def video(self):
table = PrettyTable()
table.field_names = ['ID', 'CODEC', 'QUALITY', 'BITRATE', 'FPS']
for i, j in enumerate(self.mpd_representations['video']):
table.add_row([i, j['codec'], j['quality'], j['bitrate'], j['fps']])
print('\n' + Fore.RED + 'VIDEO')
print(table)
selected_video = self.mpd_representations['video'][int(input('ID: '))]
init_url = selected_video['playback_url'] + '?range={0}-{1}'.format(selected_video['init']['first'], selected_video['init']['last'])
self.aria2c(init_url, 'init.mp4')
selected_video['pssh'] = self.extract_pssh(os.path.join(self.folders['temp'], 'init.mp4'))
os.remove(os.path.join(self.folders['temp'], 'init.mp4'))
print(Fore.YELLOW+'\nAcquiring Content License')
self.license = self.license_request(selected_video['pssh'])
print(Fore.GREEN+'License Acquired Successfully')
print(Fore.YELLOW+'\nURL:', selected_video['playback_url'])
if not self.args.keys:
output_file_name = self.movie_details['resource'][0]['metadata']['title'] + ' ' + f'[{selected_video["quality"]}] Encrypted.mp4'
print(Fore.YELLOW+'\nDownloading', output_file_name)
video_downloaded = self.aria2c(selected_video['playback_url'], output_file_name.replace(':', ''))
print(Fore.YELLOW+'\nDecrypting Video')
self.decrypt(self.license, video_downloaded, video_downloaded.replace(' Encrypted', ''))
os.remove(video_downloaded)
else:
print(Fore.GREEN + 'n\KEYS')
for key in self.license:
if key.type == 'CONTENT':
print('{}:{}'.format(key.kid.hex(), key.key.hex()))
def audio(self):
table = PrettyTable()
table.field_names = ['ID', 'CODEC', 'BITRATE', 'LANGUAGE']
for i, j in enumerate(self.mpd_representations['audio']):
table.add_row([i, j['codec'], j['bitrate'], j['language']])
print('\n' + Fore.RED +'AUDIO')
print(table)
selected_audio = input('ID: ')
if self.args.audio:
init_url = self.mpd_representations['audio'][int(selected_audio.split(',')[-1])]['playback_url']
init_url += '?range={0}-{1}'.format(self.mpd_representations['audio'][int(selected_audio.split(',')[-1])]['init']['first'], self.mpd_representations['audio'][int(selected_audio.split(',')[-1])]['init']['last'])
self.aria2c(init_url, 'init.mp4')
pssh = self.extract_pssh(os.path.join(self.folders['temp'], 'init.mp4'))
os.remove(os.path.join(self.folders['temp'], 'init.mp4'))
print(Fore.YELLOW+'\nAcquiring Content License')
self.license = self.license_request(pssh)
print(Fore.GREEN+'License Acquired Successfully')
for x in selected_audio.split(','):
x = int(x.strip())
playback_url = self.mpd_representations['audio'][x]['playback_url']
print(Fore.YELLOW+'\nURL:', playback_url)
if not self.args.keys:
output_file_name = self.movie_details['resource'][0]['metadata']['title'] + ' ' + f'[{self.mpd_representations["audio"][x]["language"]}-{self.mpd_representations["audio"][x]["codec"]}-{self.mpd_representations["audio"][x]["bitrate"]}] Encrypted.mp4'
print(Fore.YELLOW+'\nDownloading', output_file_name)
audio_downloaded = self.aria2c(playback_url, output_file_name.replace(':', ''))
self.decrypt(self.license, audio_downloaded, audio_downloaded.replace(' Encrypted', ''))
os.remove(audio_downloaded)
else:
print(Fore.GREEN + 'n\KEYS')
for key in self.license:
if key.type == 'CONTENT':
print('{}:{}'.format(key.kid.hex(), key.key.hex()))
def subtitle(self):
table = PrettyTable()
table.field_names = ['ID', 'LANGUAGE', 'FORMAT']
for i, j in enumerate(self.mpd_representations['subtitle']):
table.add_row([i, j['language'], j['format']])
print('\n' + Fore.RED +'SUBTITLE')
print(table)
selected_subtitle = input('ID: ')
for x in selected_subtitle.split(','):
x = int(x.strip())
url = self.mpd_representations['subtitle'][x]['url']
output_file_name = self.movie_details['resource'][0]['metadata']['title'] + ' ' + f'{self.mpd_representations["subtitle"][x]["language"]}-{self.mpd_representations["subtitle"][x]["format"]}'
print(Fore.YELLOW+'\nDownloading', output_file_name)
self.aria2c(url, output_file_name)
cwd = os.getcwd()
folders = {'binaries': os.path.join(cwd, 'binaries'), 'output': os.path.join(cwd, 'output'), 'temp': os.path.join(cwd, 'temp')}
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('-u', '--url', required=True)
arg_parser.add_argument('-a', '--audio', action='store_true')
arg_parser.add_argument('-k', '--keys', action='store_true')
args = arg_parser.parse_args()
if __name__ == "__main__":
movie = Main(folders, args)
movie.auth()
movie.get_movie_details()
movie.get_movie_resources()
movie.parse_movie_resources()
if not args.audio:
#movie.subtitle()
movie.video()
movie.audio()

View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@ -0,0 +1,549 @@
import datetime
import iso8601
import itertools
import re
from pyhls import protocol
ATTRIBUTELISTPATTERN = re.compile(r'''((?:[^,"']|"[^"]*"|'[^']*')+)''')
def cast_date_time(value):
return iso8601.parse_date(value)
def format_date_time(value):
return value.isoformat()
class ParseError(Exception):
def __init__(self, lineno, line):
self.lineno = lineno
self.line = line
def __str__(self):
return 'Syntax error in manifest on line %d: %s' % (self.lineno, self.line)
def parse_m3u8(content, strict=False, custom_tags_parser=None):
'''
Given a M3U8 playlist content returns a dictionary with all data found
'''
data = {
'media_sequence': 0,
'is_variant': False,
'is_endlist': False,
'is_i_frames_only': False,
'is_independent_segments': False,
'playlist_type': None,
'playlists': [],
'segments': [],
'iframe_playlists': [],
'media': [],
'keys': [],
'rendition_reports': [],
'skip': {},
'part_inf': {},
'session_data': [],
'session_keys': [],
}
state = {
'expect_segment': False,
'expect_playlist': False,
'current_key': None,
'current_segment_map': None,
}
lineno = 0
for line in string_to_lines(content):
lineno += 1
line = line.strip()
if line.startswith(protocol.ext_x_byterange):
_parse_byterange(line, state)
state['expect_segment'] = True
elif line.startswith(protocol.ext_x_targetduration):
_parse_simple_parameter(line, data, float)
elif line.startswith(protocol.ext_x_media_sequence):
_parse_simple_parameter(line, data, int)
elif line.startswith(protocol.ext_x_discontinuity_sequence):
_parse_simple_parameter(line, data, int)
elif line.startswith(protocol.ext_x_program_date_time):
_, program_date_time = _parse_simple_parameter_raw_value(line, cast_date_time)
if not data.get('program_date_time'):
data['program_date_time'] = program_date_time
state['current_program_date_time'] = program_date_time
state['program_date_time'] = program_date_time
elif line.startswith(protocol.ext_x_discontinuity):
state['discontinuity'] = True
elif line.startswith(protocol.ext_x_cue_out_cont):
_parse_cueout_cont(line, state)
state['cue_out'] = True
elif line.startswith(protocol.ext_x_cue_out):
_parse_cueout(line, state, string_to_lines(content)[lineno - 2])
state['cue_out_start'] = True
state['cue_out'] = True
elif line.startswith(protocol.ext_x_cue_in):
state['cue_in'] = True
elif line.startswith(protocol.ext_x_cue_span):
state['cue_out'] = True
elif line.startswith(protocol.ext_x_version):
_parse_simple_parameter(line, data, int)
elif line.startswith(protocol.ext_x_allow_cache):
_parse_simple_parameter(line, data)
elif line.startswith(protocol.ext_x_key):
key = _parse_key(line)
state['current_key'] = key
if key not in data['keys']:
data['keys'].append(key)
elif line.startswith(protocol.extinf):
_parse_extinf(line, data, state, lineno, strict)
state['expect_segment'] = True
elif line.startswith(protocol.ext_x_stream_inf):
state['expect_playlist'] = True
_parse_stream_inf(line, data, state)
elif line.startswith(protocol.ext_x_i_frame_stream_inf):
_parse_i_frame_stream_inf(line, data)
elif line.startswith(protocol.ext_x_media):
_parse_media(line, data, state)
elif line.startswith(protocol.ext_x_playlist_type):
_parse_simple_parameter(line, data)
elif line.startswith(protocol.ext_i_frames_only):
data['is_i_frames_only'] = True
elif line.startswith(protocol.ext_is_independent_segments):
data['is_independent_segments'] = True
elif line.startswith(protocol.ext_x_endlist):
data['is_endlist'] = True
elif line.startswith(protocol.ext_x_map):
quoted_parser = remove_quotes_parser('uri')
segment_map_info = _parse_attribute_list(protocol.ext_x_map, line, quoted_parser)
state['current_segment_map'] = segment_map_info
# left for backward compatibility
data['segment_map'] = segment_map_info
elif line.startswith(protocol.ext_x_start):
attribute_parser = {
"time_offset": lambda x: float(x)
}
start_info = _parse_attribute_list(protocol.ext_x_start, line, attribute_parser)
data['start'] = start_info
elif line.startswith(protocol.ext_x_server_control):
_parse_server_control(line, data, state)
elif line.startswith(protocol.ext_x_part_inf):
_parse_part_inf(line, data, state)
elif line.startswith(protocol.ext_x_rendition_report):
_parse_rendition_report(line, data, state)
elif line.startswith(protocol.ext_x_part):
_parse_part(line, data, state)
elif line.startswith(protocol.ext_x_skip):
_parse_skip(line, data, state)
elif line.startswith(protocol.ext_x_session_data):
_parse_session_data(line, data, state)
elif line.startswith(protocol.ext_x_session_key):
_parse_session_key(line, data, state)
elif line.startswith(protocol.ext_x_preload_hint):
_parse_preload_hint(line, data, state)
elif line.startswith(protocol.ext_x_daterange):
_parse_daterange(line, data, state)
elif line.startswith(protocol.ext_x_gap):
state['gap'] = True
# Comments and whitespace
elif line.startswith('#'):
if callable(custom_tags_parser):
custom_tags_parser(line, data, lineno)
elif line.strip() == '':
# blank lines are legal
pass
elif state['expect_segment']:
_parse_ts_chunk(line, data, state)
state['expect_segment'] = False
elif state['expect_playlist']:
_parse_variant_playlist(line, data, state)
state['expect_playlist'] = False
elif strict:
raise ParseError(lineno, line)
# there could be remaining partial segments
if 'segment' in state:
data['segments'].append(state.pop('segment'))
return data
def _parse_key(line):
params = ATTRIBUTELISTPATTERN.split(line.replace(protocol.ext_x_key + ':', ''))[1::2]
key = {}
for param in params:
name, value = param.split('=', 1)
key[normalize_attribute(name)] = remove_quotes(value)
return key
def _parse_extinf(line, data, state, lineno, strict):
chunks = line.replace(protocol.extinf + ':', '').split(',', 1)
if len(chunks) == 2:
duration, title = chunks
elif len(chunks) == 1:
if strict:
raise ParseError(lineno, line)
else:
duration = chunks[0]
title = ''
if 'segment' not in state:
state['segment'] = {}
state['segment']['duration'] = float(duration)
state['segment']['title'] = title
def _parse_ts_chunk(line, data, state):
segment = state.pop('segment')
if state.get('program_date_time'):
segment['program_date_time'] = state.pop('program_date_time')
if state.get('current_program_date_time'):
segment['current_program_date_time'] = state['current_program_date_time']
state['current_program_date_time'] += datetime.timedelta(seconds=segment['duration'])
segment['uri'] = line
segment['cue_in'] = state.pop('cue_in', False)
segment['cue_out'] = state.pop('cue_out', False)
segment['cue_out_start'] = state.pop('cue_out_start', False)
if state.get('current_cue_out_scte35'):
segment['scte35'] = state['current_cue_out_scte35']
if state.get('current_cue_out_duration'):
segment['scte35_duration'] = state['current_cue_out_duration']
segment['discontinuity'] = state.pop('discontinuity', False)
if state.get('current_key'):
segment['key'] = state['current_key']
else:
# For unencrypted segments, the initial key would be None
if None not in data['keys']:
data['keys'].append(None)
if state.get('current_segment_map'):
segment['init_section'] = state['current_segment_map']
segment['dateranges'] = state.pop('dateranges', None)
segment['gap_tag'] = state.pop('gap', None)
data['segments'].append(segment)
def _parse_attribute_list(prefix, line, atribute_parser):
params = ATTRIBUTELISTPATTERN.split(line.replace(prefix + ':', ''))[1::2]
attributes = {}
for param in params:
name, value = param.split('=', 1)
name = normalize_attribute(name)
if name in atribute_parser:
value = atribute_parser[name](value)
attributes[name] = value
return attributes
def _parse_stream_inf(line, data, state):
data['is_variant'] = True
data['media_sequence'] = None
atribute_parser = remove_quotes_parser('codecs', 'audio', 'video', 'subtitles', 'closed_captions')
atribute_parser["program_id"] = int
atribute_parser["bandwidth"] = lambda x: int(float(x))
atribute_parser["average_bandwidth"] = int
atribute_parser["frame_rate"] = float
atribute_parser["video_range"] = str
state['stream_info'] = _parse_attribute_list(protocol.ext_x_stream_inf, line, atribute_parser)
def _parse_i_frame_stream_inf(line, data):
atribute_parser = remove_quotes_parser('codecs', 'uri')
atribute_parser["program_id"] = int
atribute_parser["bandwidth"] = int
atribute_parser["average_bandwidth"] = int
atribute_parser["video_range"] = str
iframe_stream_info = _parse_attribute_list(protocol.ext_x_i_frame_stream_inf, line, atribute_parser)
iframe_playlist = {'uri': iframe_stream_info.pop('uri'),
'iframe_stream_info': iframe_stream_info}
data['iframe_playlists'].append(iframe_playlist)
def _parse_media(line, data, state):
quoted = remove_quotes_parser('uri', 'group_id', 'language', 'assoc_language', 'name', 'instream_id', 'characteristics', 'channels')
media = _parse_attribute_list(protocol.ext_x_media, line, quoted)
data['media'].append(media)
def _parse_variant_playlist(line, data, state):
playlist = {'uri': line,
'stream_info': state.pop('stream_info')}
data['playlists'].append(playlist)
def _parse_byterange(line, state):
if 'segment' not in state:
state['segment'] = {}
state['segment']['byterange'] = line.replace(protocol.ext_x_byterange + ':', '')
def _parse_simple_parameter_raw_value(line, cast_to=str, normalize=False):
param, value = line.split(':', 1)
param = normalize_attribute(param.replace('#EXT-X-', ''))
if normalize:
value = value.strip().lower()
return param, cast_to(value)
def _parse_and_set_simple_parameter_raw_value(line, data, cast_to=str, normalize=False):
param, value = _parse_simple_parameter_raw_value(line, cast_to, normalize)
data[param] = value
return data[param]
def _parse_simple_parameter(line, data, cast_to=str):
return _parse_and_set_simple_parameter_raw_value(line, data, cast_to, True)
def _parse_cueout_cont(line, state):
param, value = line.split(':', 1)
res = re.match('.*Duration=(.*),SCTE35=(.*)$', value)
if res:
state['current_cue_out_duration'] = res.group(1)
state['current_cue_out_scte35'] = res.group(2)
def _cueout_no_duration(line):
# this needs to be called first since line.split in all other
# parsers will throw a ValueError if passed just this tag
if line == protocol.ext_x_cue_out:
return (None, None)
def _cueout_elemental(line, state, prevline):
param, value = line.split(':', 1)
res = re.match('.*EXT-OATCLS-SCTE35:(.*)$', prevline)
if res:
return (res.group(1), value)
else:
return None
def _cueout_envivio(line, state, prevline):
param, value = line.split(':', 1)
res = re.match('.*DURATION=(.*),.*,CUE="(.*)"', value)
if res:
return (res.group(2), res.group(1))
else:
return None
def _cueout_duration(line):
# this needs to be called after _cueout_elemental
# as it would capture those cues incompletely
# This was added seperately rather than modifying "simple"
param, value = line.split(':', 1)
res = re.match(r'DURATION=(.*)', value)
if res:
return (None, res.group(1))
def _cueout_simple(line):
# this needs to be called after _cueout_elemental
# as it would capture those cues incompletely
param, value = line.split(':', 1)
res = re.match(r'^(\d+(?:\.\d)?\d*)$', value)
if res:
return (None, res.group(1))
def _parse_cueout(line, state, prevline):
_cueout_state = (_cueout_no_duration(line)
or _cueout_elemental(line, state, prevline)
or _cueout_envivio(line, state, prevline)
or _cueout_duration(line)
or _cueout_simple(line))
if _cueout_state:
state['current_cue_out_scte35'] = _cueout_state[0]
state['current_cue_out_duration'] = _cueout_state[1]
def _parse_server_control(line, data, state):
attribute_parser = {
"can_block_reload": str,
"hold_back": lambda x: float(x),
"part_hold_back": lambda x: float(x),
"can_skip_until": lambda x: float(x)
}
data['server_control'] = _parse_attribute_list(
protocol.ext_x_server_control, line, attribute_parser
)
def _parse_part_inf(line, data, state):
attribute_parser = {
"part_target": lambda x: float(x)
}
data['part_inf'] = _parse_attribute_list(
protocol.ext_x_part_inf, line, attribute_parser
)
def _parse_rendition_report(line, data, state):
attribute_parser = remove_quotes_parser('uri')
attribute_parser['last_msn'] = int
attribute_parser['last_part'] = int
rendition_report = _parse_attribute_list(
protocol.ext_x_rendition_report, line, attribute_parser
)
data['rendition_reports'].append(rendition_report)
def _parse_part(line, data, state):
attribute_parser = remove_quotes_parser('uri')
attribute_parser['duration'] = lambda x: float(x)
attribute_parser['independent'] = str
attribute_parser['gap'] = str
attribute_parser['byterange'] = str
part = _parse_attribute_list(protocol.ext_x_part, line, attribute_parser)
# this should always be true according to spec
if state.get('current_program_date_time'):
part['program_date_time'] = state['current_program_date_time']
state['current_program_date_time'] += datetime.timedelta(seconds=part['duration'])
part['dateranges'] = state.pop('dateranges', None)
part['gap_tag'] = state.pop('gap', None)
if 'segment' not in state:
state['segment'] = {}
segment = state['segment']
if 'parts' not in segment:
segment['parts'] = []
segment['parts'].append(part)
def _parse_skip(line, data, state):
attribute_parser = {
"skipped_segments": int
}
data['skip'] = _parse_attribute_list(protocol.ext_x_skip, line, attribute_parser)
def _parse_session_data(line, data, state):
quoted = remove_quotes_parser('data_id', 'value', 'uri', 'language')
session_data = _parse_attribute_list(protocol.ext_x_session_data, line, quoted)
data['session_data'].append(session_data)
def _parse_session_key(line, data, state):
params = ATTRIBUTELISTPATTERN.split(line.replace(protocol.ext_x_session_key + ':', ''))[1::2]
key = {}
for param in params:
name, value = param.split('=', 1)
key[normalize_attribute(name)] = remove_quotes(value)
data['session_keys'].append(key)
def _parse_preload_hint(line, data, state):
attribute_parser = remove_quotes_parser('uri')
attribute_parser['type'] = str
attribute_parser['byterange_start'] = int
attribute_parser['byterange_length'] = int
data['preload_hint'] = _parse_attribute_list(
protocol.ext_x_preload_hint, line, attribute_parser
)
def _parse_daterange(line, date, state):
attribute_parser = remove_quotes_parser('id', 'class', 'start_date', 'end_date')
attribute_parser['duration'] = float
attribute_parser['planned_duration'] = float
attribute_parser['end_on_next'] = str
attribute_parser['scte35_cmd'] = str
attribute_parser['scte35_out'] = str
attribute_parser['scte35_in'] = str
parsed = _parse_attribute_list(
protocol.ext_x_daterange, line, attribute_parser
)
if 'dateranges' not in state:
state['dateranges'] = []
state['dateranges'].append(parsed)
def string_to_lines(string):
return string.strip().splitlines()
def remove_quotes_parser(*attrs):
return dict(zip(attrs, itertools.repeat(remove_quotes)))
def remove_quotes(string):
'''
Remove quotes from string.
Ex.:
"foo" -> foo
'foo' -> foo
'foo -> 'foo
'''
quotes = ('"', "'")
if string.startswith(quotes) and string.endswith(quotes):
return string[1:-1]
return string
def normalize_attribute(attribute):
return attribute.replace('-', '_').lower().strip()
def is_url(uri):
return uri.startswith(('https://', 'http://'))

34
pyhls/protocol.py 100644
View File

@ -0,0 +1,34 @@
ext_x_targetduration = '#EXT-X-TARGETDURATION'
ext_x_media_sequence = '#EXT-X-MEDIA-SEQUENCE'
ext_x_discontinuity_sequence = '#EXT-X-DISCONTINUITY-SEQUENCE'
ext_x_program_date_time = '#EXT-X-PROGRAM-DATE-TIME'
ext_x_media = '#EXT-X-MEDIA'
ext_x_playlist_type = '#EXT-X-PLAYLIST-TYPE'
ext_x_key = '#EXT-X-KEY'
ext_x_stream_inf = '#EXT-X-STREAM-INF'
ext_x_version = '#EXT-X-VERSION'
ext_x_allow_cache = '#EXT-X-ALLOW-CACHE'
ext_x_endlist = '#EXT-X-ENDLIST'
extinf = '#EXTINF'
ext_i_frames_only = '#EXT-X-I-FRAMES-ONLY'
ext_x_byterange = '#EXT-X-BYTERANGE'
ext_x_i_frame_stream_inf = '#EXT-X-I-FRAME-STREAM-INF'
ext_x_discontinuity = '#EXT-X-DISCONTINUITY'
ext_x_cue_out = '#EXT-X-CUE-OUT'
ext_x_cue_out_cont = '#EXT-X-CUE-OUT-CONT'
ext_x_cue_in = '#EXT-X-CUE-IN'
ext_x_cue_span = '#EXT-X-CUE-SPAN'
ext_x_scte35 = '#EXT-OATCLS-SCTE35'
ext_is_independent_segments = '#EXT-X-INDEPENDENT-SEGMENTS'
ext_x_map = '#EXT-X-MAP'
ext_x_start = '#EXT-X-START'
ext_x_server_control = '#EXT-X-SERVER-CONTROL'
ext_x_part_inf = '#EXT-X-PART-INF'
ext_x_part = '#EXT-X-PART'
ext_x_rendition_report = '#EXT-X-RENDITION-REPORT'
ext_x_skip = '#EXT-X-SKIP'
ext_x_session_data = '#EXT-X-SESSION-DATA'
ext_x_session_key = '#EXT-X-SESSION-KEY'
ext_x_preload_hint = '#EXT-X-PRELOAD-HINT'
ext_x_daterange = "#EXT-X-DATERANGE"
ext_x_gap = "#EXT-X-GAP"

View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@ -0,0 +1,362 @@
import base64
import os
import time
import binascii
from google.protobuf.message import DecodeError
from google.protobuf import text_format
from pywidevine.cdm.formats import wv_proto2_pb2 as wv_proto2
from pywidevine.cdm.session import Session
from pywidevine.cdm.key import Key
from Cryptodome.Random import get_random_bytes
from Cryptodome.Random import random
from Cryptodome.Cipher import PKCS1_OAEP, AES
from Cryptodome.Hash import CMAC, SHA256, HMAC, SHA1
from Cryptodome.PublicKey import RSA
from Cryptodome.Signature import pss
from Cryptodome.Util import Padding
import logging
class Cdm:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.sessions = {}
def open_session(self, init_data_b64, device, raw_init_data = None, offline=False):
self.logger.debug("open_session(init_data_b64={}, device={}".format(init_data_b64, device))
self.logger.info("opening new cdm session")
if device.session_id_type == 'android':
# format: 16 random hexdigits, 2 digit counter, 14 0s
rand_ascii = ''.join(random.choice('ABCDEF0123456789') for _ in range(16))
counter = '01' # this resets regularly so its fine to use 01
rest = '00000000000000'
session_id = rand_ascii + counter + rest
session_id = session_id.encode('ascii')
elif device.session_id_type == 'chrome':
rand_bytes = get_random_bytes(16)
session_id = rand_bytes
else:
# other formats NYI
self.logger.error("device type is unusable")
return 1
if raw_init_data and isinstance(raw_init_data, (bytes, bytearray)):
# used for NF key exchange, where they don't provide a valid PSSH
init_data = raw_init_data
self.raw_pssh = True
else:
init_data = self._parse_init_data(init_data_b64)
self.raw_pssh = False
if init_data:
new_session = Session(session_id, init_data, device, offline)
else:
self.logger.error("unable to parse init data")
return 1
self.sessions[session_id] = new_session
self.logger.info("session opened and init data parsed successfully")
return session_id
def _parse_init_data(self, init_data_b64):
parsed_init_data = wv_proto2.WidevineCencHeader()
try:
self.logger.debug("trying to parse init_data directly")
parsed_init_data.ParseFromString(base64.b64decode(init_data_b64)[32:])
except DecodeError:
self.logger.debug("unable to parse as-is, trying with removed pssh box header")
try:
id_bytes = parsed_init_data.ParseFromString(base64.b64decode(init_data_b64)[32:])
except DecodeError:
self.logger.error("unable to parse, unsupported init data format")
return None
self.logger.debug("init_data:")
for line in text_format.MessageToString(parsed_init_data).splitlines():
self.logger.debug(line)
return parsed_init_data
def close_session(self, session_id):
self.logger.debug("close_session(session_id={})".format(session_id))
self.logger.info("closing cdm session")
if session_id in self.sessions:
self.sessions.pop(session_id)
self.logger.info("cdm session closed")
return 0
else:
self.logger.info("session {} not found".format(session_id))
return 1
def set_service_certificate(self, session_id, cert_b64):
self.logger.debug("set_service_certificate(session_id={}, cert={})".format(session_id, cert_b64))
self.logger.info("setting service certificate")
if session_id not in self.sessions:
self.logger.error("session id doesn't exist")
return 1
session = self.sessions[session_id]
message = wv_proto2.SignedMessage()
try:
message.ParseFromString(base64.b64decode(cert_b64))
except DecodeError:
self.logger.error("failed to parse cert as SignedMessage")
service_certificate = wv_proto2.SignedDeviceCertificate()
if message.Type:
self.logger.debug("service cert provided as signedmessage")
try:
service_certificate.ParseFromString(message.Msg)
except DecodeError:
self.logger.error("failed to parse service certificate")
return 1
else:
self.logger.debug("service cert provided as signeddevicecertificate")
try:
service_certificate.ParseFromString(base64.b64decode(cert_b64))
except DecodeError:
self.logger.error("failed to parse service certificate")
return 1
self.logger.debug("service certificate:")
for line in text_format.MessageToString(service_certificate).splitlines():
self.logger.debug(line)
session.service_certificate = service_certificate
session.privacy_mode = True
return 0
def get_license_request(self, session_id):
self.logger.debug("get_license_request(session_id={})".format(session_id))
self.logger.info("getting license request")
if session_id not in self.sessions:
self.logger.error("session ID does not exist")
return 1
session = self.sessions[session_id]
# raw pssh will be treated as bytes and not parsed
if self.raw_pssh:
license_request = wv_proto2.SignedLicenseRequestRaw()
else:
license_request = wv_proto2.SignedLicenseRequest()
client_id = wv_proto2.ClientIdentification()
if not os.path.exists(session.device_config.device_client_id_blob_filename):
self.logger.error("no client ID blob available for this device")
return 1
with open(session.device_config.device_client_id_blob_filename, "rb") as f:
try:
cid_bytes = client_id.ParseFromString(f.read())
except DecodeError:
self.logger.error("client id failed to parse as protobuf")
return 1
self.logger.debug("building license request")
if not self.raw_pssh:
license_request.Type = wv_proto2.SignedLicenseRequest.MessageType.Value('LICENSE_REQUEST')
license_request.Msg.ContentId.CencId.Pssh.CopyFrom(session.init_data)
else:
license_request.Type = wv_proto2.SignedLicenseRequestRaw.MessageType.Value('LICENSE_REQUEST')
license_request.Msg.ContentId.CencId.Pssh = session.init_data # bytes
if session.offline:
license_type = wv_proto2.LicenseType.Value('OFFLINE')
else:
license_type = wv_proto2.LicenseType.Value('DEFAULT')
license_request.Msg.ContentId.CencId.LicenseType = license_type
license_request.Msg.ContentId.CencId.RequestId = session_id
license_request.Msg.Type = wv_proto2.LicenseRequest.RequestType.Value('NEW')
license_request.Msg.RequestTime = int(time.time())
license_request.Msg.ProtocolVersion = wv_proto2.ProtocolVersion.Value('CURRENT')
if session.device_config.send_key_control_nonce:
license_request.Msg.KeyControlNonce = random.randrange(1, 2**31)
if session.privacy_mode:
if session.device_config.vmp:
self.logger.debug("vmp required, adding to client_id")
self.logger.debug("reading vmp hashes")
vmp_hashes = wv_proto2.FileHashes()
with open(session.device_config.device_vmp_blob_filename, "rb") as f:
try:
vmp_bytes = vmp_hashes.ParseFromString(f.read())
except DecodeError:
self.logger.error("vmp hashes failed to parse as protobuf")
return 1
client_id._FileHashes.CopyFrom(vmp_hashes)
self.logger.debug("privacy mode & service certificate loaded, encrypting client id")
self.logger.debug("unencrypted client id:")
for line in text_format.MessageToString(client_id).splitlines():
self.logger.debug(line)
cid_aes_key = get_random_bytes(16)
cid_iv = get_random_bytes(16)
cid_cipher = AES.new(cid_aes_key, AES.MODE_CBC, cid_iv)
encrypted_client_id = cid_cipher.encrypt(Padding.pad(client_id.SerializeToString(), 16))
service_public_key = RSA.importKey(session.service_certificate._DeviceCertificate.PublicKey)
service_cipher = PKCS1_OAEP.new(service_public_key)
encrypted_cid_key = service_cipher.encrypt(cid_aes_key)
encrypted_client_id_proto = wv_proto2.EncryptedClientIdentification()
encrypted_client_id_proto.ServiceId = session.service_certificate._DeviceCertificate.ServiceId
encrypted_client_id_proto.ServiceCertificateSerialNumber = session.service_certificate._DeviceCertificate.SerialNumber
encrypted_client_id_proto.EncryptedClientId = encrypted_client_id
encrypted_client_id_proto.EncryptedClientIdIv = cid_iv
encrypted_client_id_proto.EncryptedPrivacyKey = encrypted_cid_key
license_request.Msg.EncryptedClientId.CopyFrom(encrypted_client_id_proto)
else:
license_request.Msg.ClientId.CopyFrom(client_id)
if session.device_config.private_key_available:
key = RSA.importKey(open(session.device_config.device_private_key_filename).read())
session.device_key = key
else:
self.logger.error("need device private key, other methods unimplemented")
return 1
self.logger.debug("signing license request")
hash = SHA1.new(license_request.Msg.SerializeToString())
signature = pss.new(key).sign(hash)
license_request.Signature = signature
session.license_request = license_request
self.logger.debug("license request:")
for line in text_format.MessageToString(session.license_request).splitlines():
self.logger.debug(line)
self.logger.info("license request created")
self.logger.debug("license request b64: {}".format(base64.b64encode(license_request.SerializeToString())))
return license_request.SerializeToString()
def provide_license(self, session_id, license_b64):
self.logger.debug("provide_license(session_id={}, license_b64={})".format(session_id, license_b64))
self.logger.info("decrypting provided license")
if session_id not in self.sessions:
self.logger.error("session does not exist")
return 1
session = self.sessions[session_id]
if not session.license_request:
self.logger.error("generate a license request first!")
return 1
license = wv_proto2.SignedLicense()
try:
license.ParseFromString(base64.b64decode(license_b64))
except DecodeError:
self.logger.error("unable to parse license - check protobufs")
return 1
session.license = license
self.logger.debug("license:")
for line in text_format.MessageToString(license).splitlines():
self.logger.debug(line)
self.logger.debug("deriving keys from session key")
oaep_cipher = PKCS1_OAEP.new(session.device_key)
session.session_key = oaep_cipher.decrypt(license.SessionKey)
lic_req_msg = session.license_request.Msg.SerializeToString()
enc_key_base = b"ENCRYPTION\000" + lic_req_msg + b"\0\0\0\x80"
auth_key_base = b"AUTHENTICATION\0" + lic_req_msg + b"\0\0\2\0"
enc_key = b"\x01" + enc_key_base
auth_key_1 = b"\x01" + auth_key_base
auth_key_2 = b"\x02" + auth_key_base
auth_key_3 = b"\x03" + auth_key_base
auth_key_4 = b"\x04" + auth_key_base
cmac_obj = CMAC.new(session.session_key, ciphermod=AES)
cmac_obj.update(enc_key)
enc_cmac_key = cmac_obj.digest()
cmac_obj = CMAC.new(session.session_key, ciphermod=AES)
cmac_obj.update(auth_key_1)
auth_cmac_key_1 = cmac_obj.digest()
cmac_obj = CMAC.new(session.session_key, ciphermod=AES)
cmac_obj.update(auth_key_2)
auth_cmac_key_2 = cmac_obj.digest()
cmac_obj = CMAC.new(session.session_key, ciphermod=AES)
cmac_obj.update(auth_key_3)
auth_cmac_key_3 = cmac_obj.digest()
cmac_obj = CMAC.new(session.session_key, ciphermod=AES)
cmac_obj.update(auth_key_4)
auth_cmac_key_4 = cmac_obj.digest()
auth_cmac_combined_1 = auth_cmac_key_1 + auth_cmac_key_2
auth_cmac_combined_2 = auth_cmac_key_3 + auth_cmac_key_4
session.derived_keys['enc'] = enc_cmac_key
session.derived_keys['auth_1'] = auth_cmac_combined_1
session.derived_keys['auth_2'] = auth_cmac_combined_2
self.logger.debug('verifying license signature')
lic_hmac = HMAC.new(session.derived_keys['auth_1'], digestmod=SHA256)
lic_hmac.update(license.Msg.SerializeToString())
self.logger.debug("calculated sig: {} actual sig: {}".format(lic_hmac.hexdigest(), binascii.hexlify(license.Signature)))
if lic_hmac.digest() != license.Signature:
self.logger.info("license signature doesn't match - writing bin so they can be debugged")
with open("original_lic.bin", "wb") as f:
f.write(base64.b64decode(license_b64))
with open("parsed_lic.bin", "wb") as f:
f.write(license.SerializeToString())
self.logger.info("continuing anyway")
self.logger.debug("key count: {}".format(len(license.Msg.Key)))