#!/usr/bin/python3 import sys import os import base64 import json import datetime import pytz import time import uuid import hmac import hashlib import random import string from urllib.parse import urlparse from datetime import timezone, timedelta # 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 pywidevine.cdm import Cdm from pywidevine.device import Device from pywidevine.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) if challenge == "cert": challenge = "CAQ=" # Configuration WVD_PATH = './WVD.wvd' authFile = '/NowTV_auth.json' SCRIPT_DIR = os.path.abspath(os.path.dirname(__file__)) USER_AGENT = 'okhttp/4.11.0' API_USER_AGENT = 'Dalvik/2.1.0 (Linux; U; Android 14; sdk_gphone64_arm64 Build/UE1A.230829.036.A1)' PLATFORM = 'ANDROIDMOBILE' DEVICE_TYPE = 'MOBILE' MODEL = 'sdk_gphone64_arm64' VIDEO_FORMAT = 'HD' headers = { 'User-Agent': API_USER_AGENT, 'x-skyott-proposition': 'NOWOTT', 'x-skyott-provider': 'NOWTV', } token = '' territory = 'GB' language = 'en-GB' 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 generate_device_id(length=44): characters = string.ascii_uppercase + string.ascii_lowercase + string.digits return ''.join(random.choices(characters, k=length)) def calculate_signature(method, url, hdrs, payload, timestamp=None): app_id = 'NOWOTT-ANDROID-v1' signature_key = bytearray('He97trFdwMSKZBbnJGjzyPXN3Qgu2qRvh4spkmcC', 'utf-8') sig_version = '1.0' if not timestamp: timestamp = int(time.time()) if url.startswith('http'): parsed_url = urlparse(url) path = parsed_url.path else: path = url text_headers = '' for key in sorted(hdrs.keys()): if key.lower().startswith('x-skyott'): text_headers += key.lower() + ': ' + hdrs[key] + '\n' headers_md5 = hashlib.md5(text_headers.encode()).hexdigest() if isinstance(payload, str): payload = payload.encode('utf-8') payload_md5 = hashlib.md5(payload).hexdigest() to_hash = f'{method}\n{path}\n\n{app_id}\n{sig_version}\n{headers_md5}\n{timestamp}\n{payload_md5}\n' hashed = hmac.new(signature_key, to_hash.encode('utf8'), hashlib.sha1).digest() signature = base64.b64encode(hashed).decode('utf8') return f'SkyOTT client="{app_id}",signature="{signature}",timestamp="{timestamp}",version="{sig_version}"' def get_territory(): response = req.get('https://geolocation.onetrust.com/cookieconsentpub/v1/geo/location', headers={'accept': 'application/json', 'origin': 'https://www.nowtv.com', 'referer': 'https://www.nowtv.com/', 'user-agent': USER_AGENT}) return response.json()['country'] def token_request(idsession, device_id): global territory, language url = 'https://ovp.nowtv.com/auth/throttled/tokens' cookies = {'idsession': idsession} tok_headers = { 'Accept': 'application/vnd.tokens.v1+json', 'User-Agent': API_USER_AGENT, 'X-SkyOTT-Provider': 'NOWTV', 'X-SkyOTT-Proposition': 'NOWOTT', 'X-SkyOTT-Platform': PLATFORM, 'X-SkyOTT-Device': DEVICE_TYPE, 'X-SkyOTT-Territory': territory, 'X-SkyOTT-ActiveTerritory': territory, 'Content-Type': 'application/vnd.tokens.v1+json', } data = {'auth': {'authScheme': 'MESSO', 'provider': 'NOWTV', 'providerTerritory': territory, 'proposition': 'NOWOTT'}, 'device': {'type': DEVICE_TYPE, 'platform': PLATFORM, 'id': device_id, 'drmDeviceId': 'UNKNOWN'}} data = json.dumps(data) tok_headers['x-sky-signature'] = calculate_signature('POST', url, tok_headers, data) response = req.post(url, headers=tok_headers, data=data, cookies=cookies) return response.json()['userToken'] def login(): global token, territory, language print("logging in...", file=sys.stderr) auth = get_auth() if not auth or 'idsession' not in auth: print("No auth found. Please add NowTV auth to auth file.", file=sys.stderr) print("Login to NowTV in browser and copy 'idsession' cookie value", file=sys.stderr) save_auth({'idsession': 'ENTER_IDSESSION_HERE', 'device_id': generate_device_id()}) sys.exit(1) try: territory = get_territory() if territory == 'IT': language = 'it-IT' elif territory == 'DE': language = 'de-DE' else: language = 'en-GB' token = token_request(auth['idsession'], auth['device_id']) print("logged in successfully", file=sys.stderr) return token except Exception as e: print(f"Login failed: {e}", file=sys.stderr) sys.exit(1) def get_token(): global token if token: return token return login() def get_pssh_from_mpd(url): response = req.get(url, headers={'User-Agent': USER_AGENT}) try: content_protections = BeautifulSoup(response.content, features="xml").findAll('ContentProtection') for cp in content_protections: if cp.get('schemeIdUri', '').lower() == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': pssh_elem = cp.find('cenc:pssh') if pssh_elem: return pssh_elem.text except: pass return None def get_single(service_key): url = 'https://ovp.nowtv.com/video/playouts/live' single_headers = { 'User-Agent': API_USER_AGENT, 'x-skyott-usertoken': token, 'x-skyott-territory': territory, 'x-skyott-provider': 'NOWTV', 'x-skyint-requestid': str(uuid.uuid4()), 'x-skyott-proposition': 'NOWOTT', 'x-skyott-device': DEVICE_TYPE, 'x-skyott-platform': PLATFORM, 'x-skyott-activeterritory': territory, 'x-skyott-language': language, 'Content-Type': 'application/vnd.playlive.v1+json', } data = { 'device': {'model': MODEL, 'maxVideoFormat': VIDEO_FORMAT, 'supportedColourSpaces': ['SDR', 'HDR10'], 'hdcpEnabled': True, 'capabilities': [{'transport': 'DASH', 'protection': 'WIDEVINE', 'vcodec': 'H264', 'acodec': 'AAC', 'container': 'ISOBMFF'}]}, 'client': {'thirdParties': ['FREEWHEEL', 'MEDIATAILOR', 'CONVIVA', 'FRV4'], 'timeShiftEnabled': False, 'variantCapable': True}, 'parentalControlPin': None, 'personaParentalControlRating': '15', 'serviceKey': service_key } data = json.dumps(data) single_headers['x-sky-signature'] = calculate_signature('POST', url, single_headers, data) response = req.post(url, headers=single_headers, data=data) try: result = response.json() dvr_duration = result['asset']['streamVariants'][0]['dvrWindowDuration']['PT2H']['value'] urls = [e['url'].split('?')[0].replace('$dvrWindowDuration$', dvr_duration) for e in result['asset']['endpoints']] lic_url = result['protection']['licenceAcquisitionUrl'] heartbeat_url = result['events']['heartbeat']['url'] return urls, lic_url, heartbeat_url except: return None, None, None def stop_concurrency(hb_url): hb_headers = { 'X-SkyOTT-Platform': PLATFORM, 'X-SkyOTT-ActiveTerritory': territory, 'X-SkyOTT-Provider': 'NOWTV', 'X-SkyOTT-Proposition': 'NOWOTT', 'X-SkyOTT-Device': DEVICE_TYPE, 'X-SkyOTT-Territory': territory, 'X-SkyOTT-UserToken': token, 'Accept': 'application/vnd.stopstream.v1+json', 'Content-Type': 'application/vnd.stopstream.v1+json', 'User-Agent': API_USER_AGENT, } data = '{}' hb_headers['x-sky-signature'] = calculate_signature('PUT', hb_url, hb_headers, data) req.put(hb_url, headers=hb_headers, data=data) def do_cdm_internal(challenge_b64, lic_url): lic_headers = {'Origin': 'https://www.nowtv.com', 'Referer': 'https://www.nowtv.com/', 'User-Agent': USER_AGENT, 'Content-Type': 'application/octet-stream'} response = req.post(lic_url, headers=lic_headers, data=base64.b64decode(challenge_b64)) response_b64 = str(base64.b64encode(response.content), 'ascii') if response_b64.startswith('CA'): return response_b64 return None def do_cdm_external(pssh_b64, lic_url): 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) lic_headers = {'Origin': 'https://www.nowtv.com', 'Referer': 'https://www.nowtv.com/', 'User-Agent': USER_AGENT, 'Content-Type': 'application/octet-stream'} licence = req.post(lic_url, headers=lic_headers, data=challenge_data) cdm_obj.parse_license(session_id, licence.content) keys = [f"{key.kid.hex}:{key.key.hex()}" for key in cdm_obj.get_keys(session_id) if key.type != 'SIGNING'] 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': []} tz_offset = timedelta(hours=2) tz = timezone(tz_offset) now = datetime.datetime.now(tz) minute = (now.minute // 5) * 5 rounded_time = now.replace(minute=minute, second=0, microsecond=0) formatted_time = rounded_time.isoformat(timespec='minutes') ch_headers = {**headers, 'x-skyott-activeterritory': territory, 'x-skyott-device': DEVICE_TYPE, 'x-skyott-language': language, 'x-skyott-platform': PLATFORM, 'x-skyott-territory': territory} params = {'startTime': formatted_time, 'assetsPerChannelCapLimit': '4', 'playout_content_segments': 'NEWS,SHORTFORM,SPORTS,SPORTS_BONUS,SPORTS_CORE,SPORTS_ESSENTIALS,SPORTS_EVENTS,SSN', 'discovery_content_segments': 'ENTERTAINMENT,HAYU,KIDS,MOVIES,NEWS,SHORTFORM,SPORTS,SPORTS_CORE,SPORTS_ESSENTIALS,SPORTS_EVENTS,SPORTS_EVENTS_EXCLUSIVE,SSN'} response = req.get('https://web.clients.nowtv.com/bff/channel_guide', params=params, headers=ch_headers) try: data = response.json() for ch in data.get('channels', []): channel = { 'Name': ch.get('name', ch.get('title', 'Unknown')), 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={ch.get('serviceKey', '')}", 'CdmType': 'widevine', 'UseCdm': True, 'Cdm': f"id={ch.get('serviceKey', '')}", '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: service_key = id urls, lic_url, heartbeat_url = get_single(service_key) if not urls: return "error" video_url = urls[0] stop_concurrency(heartbeat_url) 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}, "LicenseUrl": lic_url } print(json.dumps(output)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "internal": try: service_key = id urls, lic_url, heartbeat_url = get_single(service_key) if lic_url: stop_concurrency(heartbeat_url) result = do_cdm_internal(challenge, lic_url) if result: print(result) else: return "error" else: return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "external": try: service_key = id urls, lic_url, heartbeat_url = get_single(service_key) if lic_url and urls: stop_concurrency(heartbeat_url) pssh_to_use = pssh if pssh else get_pssh_from_mpd(urls[0]) if pssh_to_use: keys = do_cdm_external(pssh_to_use, lic_url) 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()