#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import base64 import jwt 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') 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 = '/DirecTVGo_' + user + '.tokens' user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' def find_wv_pssh_offsets(raw): offsets = [] offset = 0 while True: offset = raw.find(b'pssh', offset) if offset == -1: break size = int.from_bytes(raw[offset-4:offset], byteorder='big') pssh_offset = offset - 4 offsets.append(raw[pssh_offset:pssh_offset+size]) offset += size return offsets def to_pssh(content): wv_offsets = find_wv_pssh_offsets(content) return [base64.b64encode(wv_offset).decode() for wv_offset in wv_offsets] def do_cdm_external(pssh_data, license_url, token): 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 = {'authorization': 'Bearer ' + token, 'origin': 'https://www.directvgo.com', 'referer': 'https://www.directvgo.com/', 'user-agent': user_agent, 'content-type': 'application/octet-stream'} licence = req.post(license_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 def get_pssh_from_mpd(url): headers = {'accept': '*/*', 'user-agent': user_agent, 'content-type': 'application/octet-stream'} response = req.get(url, headers=headers) soup = BeautifulSoup(response.content, features="xml") for cp in soup.find_all('ContentProtection'): if cp.get('schemeIdUri', '').lower() == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': pssh_tag = cp.find('cenc:pssh') if pssh_tag: return pssh_tag.text seg_templates = soup.find_all('SegmentTemplate') for st in seg_templates: init = st.get('initialization') if init: rep = soup.find('Representation') bandwidth = rep.get('bandwidth', '') rep_id = rep.get('id', '') base_url = soup.find('BaseURL') loc_parts = response.url.rsplit('/', 1)[0] if base_url and 'http' in base_url.text: init_url = base_url.text + init.replace('$Bandwidth$', bandwidth).replace('$RepresentationID$', rep_id) else: init_url = loc_parts + '/' + init.replace('$Bandwidth$', bandwidth).replace('$RepresentationID$', rep_id) init_resp = req.get(init_url, headers=headers) psshs = to_pssh(init_resp.content) return psshs[0] if psshs else None return None def get_channels(token, profile_token): headers = {'accept': 'application/json, text/plain, */*', 'authorization': 'Bearer ' + token, 'origin': 'https://www.directvgo.com', 'referer': 'https://www.directvgo.com/', 'user-agent': user_agent, 'x-app': 'dgo', 'x-client-id': 'web', 'x-client-version': '1.0', 'x-environment': 'prd', 'x-profile-token': profile_token} decoded = jwt.decode(token, options={"verify_signature": False}) channel_ids = ','.join(decoded.get('channels', [])) params = {'startTime': datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z', 'channelId': channel_ids, 'assetToken': 'false', 'language': 'es'} response = req.get('https://api.directvgo.com/contents/v2/live/schedules', headers=headers, params=params) return response.json().get('channels', []) def get_single(token, profile_token, content_id): headers = {'accept': 'application/json, text/plain, */*', 'authorization': 'Bearer ' + token, 'content-type': 'application/json', 'origin': 'https://www.directvgo.com', 'referer': 'https://www.directvgo.com/', 'user-agent': user_agent, 'x-app': 'dgo', 'x-client-id': 'web', 'x-client-version': '1.0', 'x-environment': 'prd', 'x-profile-token': profile_token} json_data = {'delay': 0, 'mobileNetwork': True, 'isLive': True, 'contentId': content_id} response = req.post('https://api.directvgo.com/entitlement/v3/playback/token/authorizer', headers=headers, json=json_data) data = response.json() url = data.get('manifest', {}).get('dash', {}).get('primary', '') if 'hdnts=' not in url: url = url.split('?')[0] license_url = data.get('drms', {}).get('widevine', [{}])[0].get('licenseUrl', '') authorization = data.get('authorization', '') return url, license_url, authorization def get_profile_token(token): headers = {'accept': 'application/json, text/plain, */*', 'authorization': token, 'origin': 'https://www.directvgo.com', 'referer': 'https://www.directvgo.com/', 'user-agent': user_agent, 'x-app': 'dgo', 'x-client-id': 'web', 'x-client-version': '1.0', 'x-environment': 'prd'} response = req.get('https://api.directvgo.com/customer/v1/profiles', headers=headers) return response.json()[0].get('profileToken', '') def do_refresh(refresh_token, business_unit, device_id): headers = {'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': 'https://www.directvgo.com', 'referer': 'https://www.directvgo.com/', 'user-agent': user_agent, 'x-app': 'dgo', 'x-business-unit-type': business_unit, 'x-client-id': 'web', 'x-client-type': 'web', 'x-client-version': '3.50.0', 'x-device-id': device_id, 'x-environment': 'prd'} json_data = {'grantType': 'refreshToken', 'refreshToken': refresh_token, 'region': business_unit.split('-')[0].lower()} response = req.post('https://sm-dgo.vrioservices.com/v3/oauth2/token', headers=headers, json=json_data) data = response.json() return data.get('id_token', ''), data.get('refresh_token', '') def login(): print("Please provide renewEntitlements, user, and deviceId manually in the tokens file", file=sys.stderr) print("Login to DirecTV Go 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)) refresh_token = auth['renewEntitlements']['refresh_token'] business_unit = auth['user']['businessUnit'] device_id = auth['deviceId'] token, new_refresh = do_refresh(refresh_token, business_unit, device_id) auth['renewEntitlements']['refresh_token'] = new_refresh json.dump(auth, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) profile_token = get_profile_token(token) except: return "error" if action == "channels": output = {'Channels': []} for c in get_channels(token, profile_token): output['Channels'].append({'Name': c['channelName'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + c['channelId'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + c['channelId'], 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in get_channels(token, profile_token): output['Events'].append({'Name': c['channelName'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + c['channelId'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + c['channelId'], '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": channel_id = id.replace('id=', '') if id.startswith('id=') else id url, license_url, authorization = get_single(token, profile_token, channel_id) pssh_data = get_pssh_from_mpd(url) output = {"Cdn": [], "ManifestUrl": url, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "LicenseUrl": license_url, "Authorization": authorization} if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) elif action == "cdm" and cdm_param == "external": channel_id = id.replace('id=', '') if id.startswith('id=') else id url, license_url, authorization = get_single(token, profile_token, channel_id) pssh_data = get_pssh_from_mpd(url) if pssh_data: for key in do_cdm_external(pssh_data, license_url, authorization): print(key) if do_action() == "error": print("Error: Please create token file with renewEntitlements, user, and deviceId", file=sys.stderr) sys.exit(1)