#!/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 = '/AmazonPrimeVideo_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 = '' base_url = '' base_url2 = '' loc_name = '' 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 set_region(region): global base_url, base_url2, loc_name if region == 'uk': loc_name = 'acbuk' base_url = 'https://www.amazon.co.uk' base_url2 = 'https://atv-ps-eu.amazon.co.uk' elif region == 'de': loc_name = 'acbde' base_url = 'https://www.amazon.de' base_url2 = 'https://atv-ps-eu.amazon.de' elif region == 'us': loc_name = 'main' base_url = 'https://www.amazon.com' base_url2 = 'https://atv-ps.amazon.com' else: loc_name = 'main-av' base_url = 'https://www.primevideo.com' base_url2 = 'https://atv-ps.primevideo.com' def refresh_x_acb(cks): headers = {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9', 'referer': 'https://www.google.com/', 'user-agent': USER_AGENT} response = req.get(base_url + '/gp/video/livetv', cookies=cks, headers=headers) response.raise_for_status() return response.cookies.get_dict().get(f'x-{loc_name}', cks.get(f'x-{loc_name}', '')) def login(): global cookies, device_id print("logging in...", file=sys.stderr) auth = get_auth() if not auth or f'ubid-{loc_name}' not in auth: print(f"No auth found. Please add AmazonPrimeVideo cookies to auth file.", file=sys.stderr) print(f"Login and copy ubid-{loc_name}, at-{loc_name}, x-{loc_name} cookies", file=sys.stderr) save_auth({f'ubid-{loc_name}': '', f'at-{loc_name}': '', f'x-{loc_name}': '', 'device_id': str(uuid.uuid4()), 'region': 'us'}) sys.exit(1) try: set_region(auth.get('region', 'us')) ck = {k: v for k, v in auth.items() if k not in ['device_id', 'region']} new_x = refresh_x_acb(ck) ck[f'x-{loc_name}'] = new_x auth[f'x-{loc_name}'] = new_x save_auth(auth) cookies = ck 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': base_url, 'Referer': base_url, 'User-Agent': USER_AGENT} params = {'deviceID': device_id, 'deviceTypeID': DEVICE_TYPE_ID, 'gascEnabled': 'false', 'uxLocale': 'en_GB', 'firmware': '1', 'playerType': 'xp', 'operatingSystemName': 'Windows', 'operatingSystemVersion': '10.0', 'deviceApplicationName': 'EdgeNext', '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': 'SINGLE_KEY', 'supportsEmbeddedTrickplay': 'true', 'daiSupportsEmbeddedTrickplay': 'true', 'daiLiveManifestType': 'patternTemplate,accumulating,live', 'ssaiSegmentInfoSupport': 'Base', 'ssaiStitchType': 'MultiPeriod', 'gdprEnabled': 'true', '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(base_url2 + '/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': base_url, 'Referer': base_url, 'User-Agent': USER_AGENT} params = {'deviceID': device_id, 'deviceTypeID': DEVICE_TYPE_ID, 'gascEnabled': 'false', 'uxLocale': 'en_GB', '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(base_url2 + '/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(): auth = get_auth() if auth: set_region(auth.get('region', 'us')) get_token() if action == "login": login() sys.exit() if action == "channels": output = {'Channels': []} headers = {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9', 'referer': 'https://www.google.com/', 'user-agent': USER_AGENT} response = req.get(base_url + '/gp/video/livetv', headers=headers, cookies=cookies) try: soup = BeautifulSoup(response.content, features='lxml') script_tags = soup.find_all('script', {'type': 'text/template'}) for s in script_tags: if 'paginationServiceToken' in s.text: init_data = json.loads(s.text) containers = init_data['props']['body'][0]['props']['landingPage']['containers'] for c in containers: for e in c.get('entities', []): if not e.get('liveInfo') or e.get('liveInfo', {}).get('status') == 'LIVE': channel = { 'Name': e.get('title', 'Unknown'), 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={e.get('playbackAction', {}).get('channelId', e.get('impressionId', ''))}", 'CdmType': 'playready', 'UseCdm': True, 'Cdm': f"id={e.get('playbackAction', {}).get('channelId', e.get('impressionId', ''))}", 'Video': 'best', 'OnDemand': True, 'SpeedUp': True, } output['Channels'].append(channel) break 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()