#!/usr/bin/python3 import sys import os import o11 import base64 import json import datetime import pytz import requests import hmac import hashlib import time from urllib.parse import urlparse from pywidevine.cdm import Cdm from pywidevine.device import Device from pywidevine.pssh import PSSH from bs4 import BeautifulSoup WVD_PATH = './WVD.wvd' user = o11.parse_params(sys.argv, 'user') password = o11.parse_params(sys.argv, 'password') device_param = o11.parse_params(sys.argv, 'device') pin = o11.parse_params(sys.argv, 'pin') country = o11.parse_params(sys.argv, 'country') ssoToken = o11.parse_params(sys.argv, 'ssoToken') 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') o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" authFile = '/SkyGo_' + user + '.tokens' proxies = {"http": proxy, "https": proxy} if proxy else {} platform = 'ANDROID' device_type = 'MOBILE' user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36' def calculate_signature(method, url, headers, payload, timestamp=None): app_id = 'SKYGO-ANDROID-v2' signature_key = bytearray('jM3CCqNAAQBY9VDeVd6rjttYc2cYDpqnsvNxuTfH', '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(headers.keys()): if key.lower().startswith('x-skyott'): text_headers += key + ': ' + headers[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 = ('{method}\n{path}\n{response_code}\n{app_id}\n{version}\n{headers_md5}\n' '{timestamp}\n{payload_md5}\n').format(method=method, path=path, response_code='', app_id=app_id, version=sig_version, headers_md5=headers_md5, timestamp=timestamp, payload_md5=payload_md5) hashed = hmac.new(signature_key, to_hash.encode('utf8'), hashlib.sha1).digest() signature = base64.b64encode(hashed).decode('utf8') return 'SkyOTT client="{}",signature="{}",timestamp="{}",version="{}"'.format(app_id, signature, timestamp, sig_version) def get_pssh_from_mpd(manifest_url): response = req.get(manifest_url) if response.status_code != 200: return None psshs = BeautifulSoup(response.content, features="xml").findAll('cenc:pssh') if len(psshs) > 1: return psshs[1].text elif len(psshs) > 0: return psshs[0].text return None def do_cdm_external(pssh_data, license_url): """Extract Widevine keys using pywidevine library""" pssh_obj = PSSH(pssh_data) 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 = {'Content-Type': 'application/octet-stream'} licence = req.post(license_url, headers=lic_headers, data=challenge_data) try: licence.raise_for_status() cdm_obj.parse_license(session_id, licence.content) keys = [] for key in cdm_obj.get_keys(session_id): if key.type != 'SIGNING': keys.append(f"{key.kid.hex}:{key.key.hex()}") cdm_obj.close(session_id) return keys except Exception as e: print(f'License request failed: {e}', file=sys.stderr) return [] def get_region(): response = req.get('https://init.sky.com/') data = response.json() return data['geoip']['country-code'] def get_token(sso_token, country_code): token_url = 'https://p.sky.com/auth/tokens' headers = { 'accept': 'application/vnd.tokens.v1+json', 'content-type': 'application/vnd.tokens.v1+json', 'x-skyott-device': device_type, 'x-skyott-platform': platform, 'x-skyott-proposition': 'SKYQ', 'x-skyott-provider': 'SKY', 'x-skyott-territory': country_code } post_data = { "auth": { "authScheme": "MESSO", "authToken": sso_token, "provider": "SKY", "providerTerritory": country_code, "proposition": "SKYQ" }, "device": { "type": device_type, "platform": platform, "id": 'Z-sKxKApCe7c3dBMGAYtKU8NmWKDcWrCKobKpnVTLqc', "drmDeviceId": "UNKNOWN" } } post_data_str = json.dumps(post_data) headers['x-sky-signature'] = calculate_signature('POST', token_url, headers, post_data_str) response = req.post(token_url, headers=headers, data=post_data_str) data = response.json() return data['userToken'] def get_channels(country_code): headers = { 'X-SkyOTT-Territory': country_code, 'X-SkyOTT-Proposition': 'SKYQ', 'X-SkyOTT-Provider': 'SKY', 'X-SkyOTT-Device': device_type, 'X-SkyOTT-Platform': platform, 'X-SkyOTT-Application': 'SKYQ/23.7.0', 'User-Agent': f'skyq:android:23.7.0:{country_code}' } response = req.get('https://atlantis.epgsky.com/as/services/4104/32', headers=headers) data = response.json() channels = [] services = data.get('services', []) for s in services: if 'OTT' in s.get('servicetypes', []): channels.append(s) return channels def live_request(service_key, country_code, user_token): live_url = 'https://p.sky.com/video/playouts/live' headers = { 'x-skyott-device': device_type, 'x-skyott-pinoverride': 'true', 'x-skyott-platform': platform, 'x-skyott-proposition': 'SKYQ', 'x-skyott-provider': 'SKY', 'x-skyott-territory': country_code, 'x-skyott-usertoken': user_token, 'Content-Type': 'application/vnd.playlive.v1+json', } post_data = { "serviceKey": service_key, "device": { "capabilities": [ {"transport": "DASH", "protection": "WIDEVINE", "vcodec": "H265", "acodec": "EAC3", "container": "ISOBMFF"}, {"transport": "DASH", "protection": "WIDEVINE", "vcodec": "H264", "acodec": "EAC3", "container": "ISOBMFF"}, {"transport": "DASH", "protection": "WIDEVINE", "vcodec": "H265", "acodec": "AAC", "container": "ISOBMFF"}, {"transport": "DASH", "protection": "WIDEVINE", "vcodec": "H264", "acodec": "AAC", "container": "ISOBMFF"}, ], "model": "SM-G955F", "maxVideoFormat": "UHD", "hdcpEnabled": 'false', "supportedColourSpaces": ["DV", "HDR10", "SDR"] }, "client": {"thirdParties": ["COMSCORE", "CONVIVA", "FRIENDS"]} } post_data_str = json.dumps(post_data) headers['x-sky-signature'] = calculate_signature('POST', live_url, headers, post_data_str) response = req.post(live_url, headers=headers, data=post_data_str) data = response.json() manifest_url = data.get('asset', {}).get('endpoints', [{}])[0].get('url', '').split('?')[0] license_url = data.get('protection', {}).get('licenceAcquisitionUrl') return manifest_url, license_url def login(): print("Saving tokens...", file=sys.stderr) country_code = country if country else get_region() if not ssoToken: print("Please provide ssoToken parameter (skyCEsidismesso01 cookie from browser)", file=sys.stderr) sys.exit(1) user_token = get_token(ssoToken, country_code) auth_data = {'ssoToken': ssoToken, 'userToken': user_token, 'countryCode': country_code} json.dump(auth_data, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) print("Tokens saved successfully", file=sys.stderr) def do_action(): if action == "login": login() sys.exit() try: auth = json.load(open(os.path.abspath(os.path.dirname(__file__)) + authFile)) sso_token = auth['ssoToken'] country_code = auth['countryCode'] user_token = get_token(sso_token, country_code) auth['userToken'] = user_token json.dump(auth, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) except: return "error" if action == "channels": output = {'Channels': []} channels = get_channels(country_code) for chan in channels: channel = { 'Name': chan['t'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(chan['sid']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['sid']), 'Video': 'best', 'OnDemand': True, 'SpeedUp': True } output['Channels'].append(channel) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} channels = get_channels(country_code) for chan in channels: event = { 'Name': chan['t'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(chan['sid']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['sid']), 'Video': 'best', 'Autostart': True, 'Start': int(datetime.datetime.now(pytz.UTC).timestamp()), 'End': int((datetime.datetime.now(pytz.UTC) + datetime.timedelta(hours=4)).timestamp()) } output['Events'].append(event) print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": service_key = id try: manifest_url, license_url = live_request(service_key, country_code, user_token) pssh_data = get_pssh_from_mpd(manifest_url) output = { "Cdn": [], "ManifestUrl": manifest_url, "Headers": { "Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent} }, "Heartbeat": { "Url": '', "Params": '', "PeriodMs": 5*60*1000 }, "LicenseUrl": license_url } if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) except Exception as e: print(f"Error getting manifest: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "internal": lic_url = o11.parse_params(sys.argv, 'licenseUrl') try: lic_headers = {'Content-Type': 'application/octet-stream'} response = req.post(lic_url, headers=lic_headers, data=base64.b64decode(challenge)) response_b64 = str(base64.b64encode(response.content), 'ascii') if response_b64.startswith('CA'): print(response_b64) else: print(response.text, file=sys.stderr) return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "external": service_key = id try: manifest_url, license_url = live_request(service_key, country_code, user_token) pssh_data = get_pssh_from_mpd(manifest_url) if pssh_data and license_url: keys = do_cdm_external(pssh_data, license_url) for key in keys: print(key) else: print("No PSSH or license URL found", file=sys.stderr) 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()