#!/usr/bin/python3 import sys import os import o11 import base64 import json import datetime import pytz import requests import xmltodict from pywidevine.cdm import Cdm from pywidevine.device import Device from pywidevine.pssh import PSSH WVD_PATH = './WVD.wvd' user = o11.parse_params(sys.argv, 'user') password = o11.parse_params(sys.argv, 'password') session_id_param = o11.parse_params(sys.argv, 'sessionId') access_token_param = o11.parse_params(sys.argv, 'accessToken') 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 = '/Digitalb_' + user + '.tokens' proxies = {"http": proxy, "https": proxy} if proxy else {} user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36' def do_cdm_external(pssh_data): """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 = { 'accept': '*/*', 'user-agent': user_agent, } licence = req.post('https://prd-digitalb.spectar.tv/drm.php/widevine', 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 pssh_kid_grabber(mpd_data): try: for adaptations in mpd_data['MPD']['Period']['AdaptationSet']: if adaptations.get('@mimeType') == 'video/mp4': for protections in adaptations['Representation'][0]['ContentProtection']: if protections.get('@schemeIdUri', '').lower() == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': return protections['cenc:pssh']['#text'] except: pass try: for adaptations in mpd_data['MPD']['Period']['AdaptationSet']: if adaptations.get('@contentType') == 'video': for protections in adaptations['ContentProtection']: if protections.get('@schemeIdUri', '').lower() == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': return protections.get('cenc:pssh') except: pass return None def get_channels(session_id_val, access_token): headers = {'accept': '*/*', 'user-agent': user_agent} url = f'https://prd-digitalb.spectar.tv/client_api.php/channel/list/session_id/{session_id_val}/access_token/{access_token}/format/json' response = req.get(url, headers=headers) return response.json() def login(): print("Saving tokens...", file=sys.stderr) if not session_id_param or not access_token_param: print("Please provide sessionId and accessToken parameters from browser DevTools", file=sys.stderr) sys.exit(1) auth_data = {'sessionId': session_id_param, 'accessToken': access_token_param} 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)) session_id_val = auth['sessionId'] access_token = auth['accessToken'] except: return "error" if action == "channels": output = {'Channels': []} channels = get_channels(session_id_val, access_token) for chan in channels: mpd_url = chan.get('url', '').split('&profile_id')[0] channel = { 'Name': chan['name'], 'Mode': "live", 'SessionManifest': False, 'ManifestScript': 'id=' + str(chan['id']) + '&url=' + mpd_url, 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['id']) + '&url=' + mpd_url, 'Video': 'best', 'OnDemand': True, 'SpeedUp': True } output['Channels'].append(channel) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} channels = get_channels(session_id_val, access_token) for chan in channels: mpd_url = chan.get('url', '').split('&profile_id')[0] event = { 'Name': chan['name'], 'Mode': "live", 'SessionManifest': False, 'ManifestScript': 'id=' + str(chan['id']) + '&url=' + mpd_url, 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['id']) + '&url=' + mpd_url, '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": mpd_url = o11.parse_params(sys.argv, 'url') if not mpd_url and '&url=' in id: parts = id.split('&url=') mpd_url = parts[1] if len(parts) > 1 else '' try: output = { "Cdn": [], "ManifestUrl": mpd_url, "Headers": { "Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent} }, "Heartbeat": {"Url": '', "Params": '', "PeriodMs": 5*60*1000} } # Get PSSH from MPD headers = {'accept': '*/*', 'user-agent': user_agent} resp = req.get(mpd_url, headers=headers) mpd_data = json.loads(json.dumps(xmltodict.parse(resp.text))) pssh_data = pssh_kid_grabber(mpd_data) 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": try: lic_headers = {'accept': '*/*', 'user-agent': user_agent} response = req.post('https://prd-digitalb.spectar.tv/drm.php/widevine', 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": mpd_url = o11.parse_params(sys.argv, 'url') if not mpd_url and '&url=' in id: parts = id.split('&url=') mpd_url = parts[1] if len(parts) > 1 else '' try: headers = {'accept': '*/*', 'user-agent': user_agent} resp = req.get(mpd_url, headers=headers) mpd_data = json.loads(json.dumps(xmltodict.parse(resp.text))) pssh_data = pssh_kid_grabber(mpd_data) if pssh_data: keys = do_cdm_external(pssh_data) for key in keys: print(key) else: print("No PSSH found in manifest", 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()