#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import base64 import m3u8 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') 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_param = o11.parse_params(sys.argv, 'cdm') challenge = o11.parse_params(sys.argv, 'challenge') o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" authFile = '/DirectTV_' + user + '.tokens' 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 hex_to_base64(hex_string): return base64.b64encode(bytes.fromhex(hex_string)).decode() def activate_device(token, activation_token): pssh_obj = PSSH('AAAAOHBzc2gAAAAA7e+LqXnWSs6jyCfc1R0h7QAAABgSEE4LgWhhEsTwiC+3/sTT5ZRI88aJmwYAAAA0cHNzaAEAAAAQd+/swLJNAqzjPB5S4vtLAAAAAU4LgWhhEsTwiC+3/sTT5ZQAAAAA') 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) payload = base64.b64encode(challenge_data).decode() headers = {'Accept': 'application/json', 'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json', 'Origin': 'https://stream.directv.com', 'Referer': 'https://stream.directv.com/', 'User-Agent': user_agent} data = json.dumps({'activationToken': hex_to_base64(activation_token), 'activationChallenge': payload}).replace(' ', '') response = req.post('https://api.cld.dtvce.com/rights/management/mdrm/vgemultidrm/v1/widevine/activate', headers=headers, data=data) cdm_obj.close(session_id) return response.json().get('identityCookie', '') def do_cdm_external(pssh_data, feed_id, token, authorization_token, identity_cookie): 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) payload = base64.b64encode(challenge_data).decode() lic_headers = {'accept': 'application/json', 'authorization': 'Bearer ' + token, 'content-type': 'application/json', 'origin': 'https://stream.directv.com', 'referer': 'https://stream.directv.com/', 'user-agent': user_agent} data = json.dumps({'contentID': feed_id, 'contentType': '2', 'identityCookie': identity_cookie, 'authorizationToken': authorization_token, 'licenseChallenge': payload}).replace(' ', '') licence = req.post('https://api.cld.dtvce.com/rights/management/mdrm/vgemultidrm/v1/widevine/license', headers=lic_headers, data=data) lic_data = licence.json() cdm_obj.parse_license(session_id, base64.b64decode(lic_data['licenseData'][0])) 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 def get_pssh_from_m3u8(manifest_url, fallback_url): try: manifest = m3u8.load(manifest_url) url = manifest_url except: manifest = m3u8.load(fallback_url) url = fallback_url psshs = [] for key in manifest.session_keys: if key and 'base64' in key.uri: pssh = key.uri.split(',')[-1] if len(pssh) > 0 and pssh not in psshs: psshs.append(pssh) return psshs, url def get_channels(token): headers = {'accept': '*/*', 'authorization': 'Bearer ' + token, 'origin': 'https://stream.directv.com', 'referer': 'https://stream.directv.com/', 'user-agent': user_agent} params = {'sort': 'OrdCh%3DASC', 'include4K': 'false', 'is4KCompatible': 'false', 'isFtue': 'true'} response = req.get('https://api.cld.dtvce.com/discovery/metadata/channel/v5/service/allchannels', params=params, headers=headers) channels = response.json().get('channelInfoList', []) return [c for c in channels if c.get('augmentation', {}).get('constraints', {}).get('isLiveStreamEnabled')] def get_single(content_id, token): headers = {'accept': '*/*', 'authorization': 'Bearer ' + token, 'content-type': 'application/json', 'origin': 'https://stream.directv.com', 'referer': 'https://stream.directv.com/', 'user-agent': user_agent} response = req.get(f'https://api.cld.dtvce.com/right/authorization/channel/v1?ccid={content_id}&clientContext=dmaID:630_0,deviceProximity:OOH&proximity=O&timeShiftEnabled=true&daiEnabled=true&reserveCTicket=true&dualManifest=false&abrEnabled=true', headers=headers) data = response.json() return data['playbackData']['streamURL'], data['playbackData']['fallbackStreamUrl'], data['dRights']['playToken'], data['contentRef'] def do_refresh(id_token, refresh_token): headers = {'accept': '*/*', 'content-type': 'application/x-www-form-urlencoded;charset=UTF-8', 'origin': 'https://stream.directv.com', 'referer': 'https://stream.directv.com/', 'user-agent': user_agent} params = {'clientID': 'UNIFIED_DTV_WEB'} data = {'clientMake': '0', 'clientModel': ' desktop', 'refresh_token': refresh_token, 'haloCToken': id_token, 'reqParams': 'ACTIVATIONTOKEN'} response = req.post('https://api.cld.dtvce.com/authn-refreshgo/v3/refresh', params=params, headers=headers, data=data) resp_data = response.json() return resp_data['refresh_token'], resp_data['access_token'], resp_data['valuePairs']['activationToken'] def login(): print("Please provide tToken and refreshToken manually in the tokens file", file=sys.stderr) print("Login to DirectTV website, go to DevTools -> Application -> LocalStorage and copy values", file=sys.stderr) sys.exit(1) def do_action(): if action == "login": login() sys.exit() try: auth = json.load(open(os.path.abspath(os.path.dirname(__file__)) + authFile)) id_token = auth['tToken'] refresh_token = auth['refreshToken'] identity_cookie = auth.get('identityCookie', '') refresh_token, token, activation_token = do_refresh(id_token, refresh_token) if not identity_cookie: identity_cookie = activate_device(token, activation_token) auth['refreshToken'] = refresh_token auth['identityCookie'] = identity_cookie json.dump(auth, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) except: return "error" if action == "channels": output = {'Channels': []} for c in get_channels(token): output['Channels'].append({'Name': c['channelName'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'ccid=' + c['ccid'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'ccid=' + c['ccid'], 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in get_channels(token): output['Events'].append({'Name': c['channelName'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'ccid=' + c['ccid'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'ccid=' + c['ccid'], 'Video': 'best', 'Autostart': True, 'Start': int(datetime.datetime.now(pytz.UTC).timestamp()), 'End': int((datetime.datetime.now(pytz.UTC) + datetime.timedelta(hours=4)).timestamp())}) print(json.dumps(output, indent=2)) elif action == "manifest": content_id = id.replace('ccid=', '') if id.startswith('ccid=') else id url, fallback_url, authorization_token, feed_id = get_single(content_id, token) psshs, manifest_url = get_pssh_from_m3u8(url, fallback_url) output = {"Cdn": [], "ManifestUrl": manifest_url, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "FeedId": feed_id, "AuthorizationToken": authorization_token} if psshs: output['Pssh'] = psshs[0] print(json.dumps(output)) elif action == "cdm" and cdm_param == "external": content_id = id.replace('ccid=', '') if id.startswith('ccid=') else id url, fallback_url, authorization_token, feed_id = get_single(content_id, token) psshs, manifest_url = get_pssh_from_m3u8(url, fallback_url) all_keys = [] for p in psshs: all_keys += do_cdm_external(p, feed_id, token, authorization_token, identity_cookie) for key in list(set(all_keys)): print(key) if do_action() == "error": print("Error: Please create token file with tToken and refreshToken", file=sys.stderr) sys.exit(1)