#!/usr/bin/python3 import sys import os import base64 import json import datetime import pytz import uuid # Add parent directory to path for o11 import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import o11 from bs4 import BeautifulSoup from pyplayready.cdm import Cdm from pyplayready.device import Device from pyplayready.system.pssh import PSSH # Parse command line parameters user = o11.parse_params(sys.argv, 'user') password = o11.parse_params(sys.argv, 'password') device = o11.parse_params(sys.argv, 'device') pin = o11.parse_params(sys.argv, 'pin') id = o11.parse_params(sys.argv, 'id') action = o11.parse_params(sys.argv, 'action') bind = o11.parse_params(sys.argv, 'bind') proxy = o11.parse_params(sys.argv, 'proxy') doh = o11.parse_params(sys.argv, 'doh') worker = o11.parse_params(sys.argv, 'worker') cdm = o11.parse_params(sys.argv, 'cdm') drm = o11.parse_params(sys.argv, 'drm') kid = o11.parse_params(sys.argv, 'kid') pssh = o11.parse_params(sys.argv, 'pssh') challenge = o11.parse_params(sys.argv, 'challenge') heartbeaturl = o11.parse_params(sys.argv, 'heartbeaturl') heartbeatparams = o11.parse_params(sys.argv, 'heartbeatparams') # Session setup o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) # Configuration WVD_PATH = './WVD.wvd' authFile = '/PrimeVideo_cookies.json' SCRIPT_DIR = os.path.abspath(os.path.dirname(__file__)) USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36' DEVICE_TYPE_ID = 'AOAGZA014O5RE' cookies = {} device_id = '' def get_auth(): try: return json.load(open(SCRIPT_DIR + authFile)) except: return None def save_auth(auth_data): json.dump(auth_data, open(SCRIPT_DIR + authFile, 'w'), indent=2) def refresh_x_main_av(cks): headers = {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8', 'referer': 'https://www.google.com/', 'user-agent': USER_AGENT} response = req.get('https://www.primevideo.com/livetv', cookies=cks, headers=headers) response.raise_for_status() return response.cookies.get_dict().get('x-main-av', cks.get('x-main-av', '')) def login(): global cookies, device_id print("logging in...", file=sys.stderr) auth = get_auth() if not auth or 'ubid-main-av' not in auth: print("No auth found. Please add PrimeVideo cookies to auth file.", file=sys.stderr) print("Login and copy ubid-main-av, at-main-av, x-main-av cookies", file=sys.stderr) save_auth({'ubid-main-av': '', 'at-main-av': '', 'x-main-av': '', 'device_id': str(uuid.uuid4())}) sys.exit(1) try: auth['x-main-av'] = refresh_x_main_av(auth) save_auth(auth) cookies = auth device_id = auth.get('device_id', str(uuid.uuid4())) print("logged in successfully", file=sys.stderr) return cookies except Exception as e: print(f"Login failed: {e}", file=sys.stderr) sys.exit(1) def get_token(): global cookies if cookies: return cookies return login() def get_pssh_from_mpd(url): response = req.get(url, headers={'User-Agent': USER_AGENT}) content_protections = BeautifulSoup(response.content, features="xml").findAll('ContentProtection') for cp in content_protections: if cp.get('schemeIdUri', '').lower() == 'urn:uuid:9a04f079-9840-4286-ab92-e65be0885f95': pssh_elem = cp.find('cenc:pssh') if pssh_elem: return pssh_elem.text return None def get_single(channel_id): headers = {'Accept': '*/*', 'Origin': 'https://www.primevideo.com', 'Referer': 'https://www.primevideo.com/', 'User-Agent': USER_AGENT} params = {'deviceID': device_id, 'deviceTypeID': DEVICE_TYPE_ID, 'gascEnabled': 'true', 'firmware': '1', 'playerType': 'xp', 'operatingSystemName': 'Windows', 'operatingSystemVersion': '10.0', 'deviceApplicationName': 'Chrome', 'asin': channel_id, 'consumptionType': 'Streaming', 'desiredResources': 'PlaybackUrls,CuepointPlaylist,SubtitleUrls,ForcedNarratives,TrickplayUrls,TransitionTimecodes,PlaybackSettings,CatalogMetadata,LiveSchedule,LinearRestrictions,SyeUrlsV2,XRayMetadata', 'resourceUsage': 'CacheResources', 'videoMaterialType': 'LiveStreaming', 'displayWidth': '2560', 'displayHeight': '1440', 'supportsVariableAspectRatio': 'true', 'supportsEmbeddedTimedTextForVod': 'true', 'deviceProtocolOverride': 'Https', 'vodStreamSupportOverride': 'Auxiliary', 'deviceStreamingTechnologyOverride': 'DASH', 'deviceDrmOverride': 'CENC', 'deviceAdInsertionTypeOverride': 'SSAI', 'deviceHdrFormatsOverride': 'None', 'deviceVideoCodecOverride': 'H264', 'deviceVideoQualityOverride': 'HD', 'deviceBitrateAdaptationsOverride': 'CVBR,CBR', 'supportsEmbeddedTrickplayForVod': 'false', 'audioTrackId': 'all', 'languageFeature': 'MLFv2', 'liveManifestType': 'patternTemplate,accumulating,live', 'supportedDRMKeyScheme': 'DUAL_KEY', 'supportsEmbeddedTrickplay': 'true', 'daiSupportsEmbeddedTrickplay': 'true', 'daiLiveManifestType': 'patternTemplate,accumulating,live', 'ssaiSegmentInfoSupport': 'Base', 'ssaiStitchType': 'MultiPeriod', 'gdprEnabled': 'false', 'subtitleFormat': 'TTMLv2', 'playbackSettingsFormatVersion': '1.0.0', 'titleDecorationScheme': 'primary-content', 'liveScheduleConfigToken': 'html5-player-1', 'linearRestrictionStart': '-120', 'linearRestrictionEnd': '120', 'xrayToken': 'XRAY_WEB_2023_V2', 'xrayPlaybackMode': 'playback', 'xrayDeviceClass': 'normal'} response = req.post('https://atv-ps.primevideo.com/cdp/catalog/GetPlaybackResources', params=params, cookies=cookies, headers=headers) try: data = response.json() url_sets = data['playbackUrls']['urlSets'] _, set1 = next(iter(url_sets.items())) return set1['urls']['manifest']['url'] except: return None def do_cdm_external(pssh_b64, channel_id): try: pssh_obj = PSSH(pssh_b64) device_obj = Device.load(WVD_PATH) cdm_obj = Cdm.from_device(device_obj) session_id = cdm_obj.open() challenge_data = cdm_obj.get_license_challenge(session_id, pssh_obj.wrm_headers[0]) headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Origin': 'https://www.primevideo.com', 'Referer': 'https://www.primevideo.com/', 'User-Agent': USER_AGENT} params = {'deviceID': device_id, 'deviceTypeID': DEVICE_TYPE_ID, 'gascEnabled': 'true', 'marketplaceID': 'ART4WZ8MWBX2Y', 'firmware': '1', 'playerType': 'xp', 'operatingSystemName': 'Windows', 'operatingSystemVersion': '10.0', 'deviceApplicationName': 'EdgeNext', 'asin': channel_id, 'consumptionType': 'Streaming', 'desiredResources': 'PlayReadyLicense', 'resourceUsage': 'ImmediateConsumption', 'videoMaterialType': 'LiveStreaming', 'displayWidth': '2560', 'displayHeight': '1440', 'supportsVariableAspectRatio': 'true', 'supportsEmbeddedTimedTextForVod': 'true', 'deviceProtocolOverride': 'Https', 'vodStreamSupportOverride': 'Auxiliary', 'deviceStreamingTechnologyOverride': 'DASH', 'deviceDrmOverride': 'CENC', 'deviceAdInsertionTypeOverride': 'SSAI', 'deviceHdrFormatsOverride': 'None', 'deviceVideoCodecOverride': 'H264', 'deviceVideoQualityOverride': 'HD', 'deviceBitrateAdaptationsOverride': 'CVBR,CBR', 'supportsEmbeddedTrickplayForVod': 'false'} data = {'playReadyChallenge': base64.b64encode(challenge_data.encode('utf-8')).decode('utf-8')} licence = req.post('https://atv-ps.primevideo.com/cdp/catalog/GetPlaybackResources', params=params, cookies=cookies, headers=headers, data=data) lic_data = licence.json() cdm_obj.parse_license(session_id, base64.b64decode(lic_data['playReadyLicense']['encodedLicenseResponse']).decode('utf-8')) keys = [f"{key.key_id.hex}:{key.key.hex()}" for key in cdm_obj.get_keys(session_id)] cdm_obj.close(session_id) return keys except Exception as e: print(f'CDM external failed: {e}', file=sys.stderr) return None def do_action(): get_token() if action == "login": login() sys.exit() if action == "channels": output = {'Channels': []} headers = {'x-requested-with': 'WebSPA', 'Accept': 'application/json', 'Referer': 'https://www.primevideo.com/livetv', 'x-purpose': 'navigation', 'User-Agent': USER_AGENT} params = {'dvWebSPAClientVersion': '1.0.117587.0'} response = req.get('https://www.primevideo.com/livetv', params=params, headers=headers, cookies=cookies) try: data = response.json() landing_page = data['page'][0]['assembly']['body'][0]['props']['landingPage'] containers = landing_page['containers'] for c in containers: if 'containerType' in c and c['containerType'] == 'EpgGroup': if 'entities' in c: for e in c['entities']: if 'entitlementCues' in e and e['entitlementCues'].get('entitlementType') == 'Entitled': channel = { 'Name': e.get('title', 'Unknown'), 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={e.get('impressionId', '')}", 'CdmType': 'playready', 'UseCdm': True, 'Cdm': f"id={e.get('impressionId', '')}", 'Video': 'best', 'OnDemand': True, 'SpeedUp': True, } output['Channels'].append(channel) elif 'containerType' in c and c['containerType'] == 'BeardSupportedCarousel': if 'entities' in c: for e in c['entities']: if ('liveInfo' in e and e['liveInfo']['status'].lower() == 'live') and 'entitlementCues' in e and e['entitlementCues'].get('entitlementType') == 'Entitled': channel = { 'Name': e.get('title', 'Unknown'), 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={e.get('impressionId', '')}", 'CdmType': 'playready', 'UseCdm': True, 'Cdm': f"id={e.get('impressionId', '')}", 'Video': 'best', 'OnDemand': True, 'SpeedUp': True, } output['Channels'].append(channel) print(json.dumps(output, indent=2)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "events": output = {'Events': []} print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": try: channel_id = id video_url = get_single(channel_id) if not video_url: return "error" output = { "Cdn": [{"Name": "default", "ManifestUrl": video_url}], "ManifestUrl": video_url, "Headers": {"Manifest": {'User-Agent': USER_AGENT}, "Media": {'User-Agent': USER_AGENT}}, "Heartbeat": {"Url": '', "Params": '', "PeriodMs": 5*60*1000} } print(json.dumps(output)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "external": try: channel_id = id video_url = get_single(channel_id) if video_url: pssh_to_use = pssh if pssh else get_pssh_from_mpd(video_url) if pssh_to_use: keys = do_cdm_external(pssh_to_use, channel_id) if keys: for key in keys: print(key) else: return "error" else: return "error" else: return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" else: print("invalid action: " + action, file=sys.stderr) if do_action() == "error": login() do_action()