#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import urllib.parse 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 = '/F1TV_' + user + '.tokens' user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36' def do_cdm_external(pssh_data, lic_url, lic_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 = {'entitlementtoken': lic_token, 'origin': 'https://f1tv.formula1.com', 'user-agent': user_agent, 'content-type': 'application/x-www-form-urlencoded'} 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 def get_pssh_from_mpd(url): headers = {'User-Agent': user_agent} response = req.get(url, headers=headers) for cp in BeautifulSoup(response.content, features="xml").find_all('ContentProtection'): if cp.get('schemeIdUri') == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': pssh_tag = cp.find('cenc:pssh') if pssh_tag: return pssh_tag.text return None def get_events(): headers = {'x-f1-device-info': 'device=web;screen=browser;os=windows;browser=chrome;browserVersion=139.0.0.0;osVersion=10;appVersion=release-R45.0.2;playerVersion=8.212.0', 'user-agent': user_agent} response = req.get('https://f1tv.formula1.com/2.0/R/ENG/WEB_DASH/ALL/PAGE/395/REG/14', headers=headers) data = response.json() r = [] for c in data.get('resultObj', {}).get('containers', []): if 'retrieveItems' in c and 'resultObj' in c['retrieveItems'] and 'containers' in c['retrieveItems']['resultObj']: for c2 in c['retrieveItems']['resultObj']['containers']: if 'events' in c2: for e in c2['events']: if 'metadata' in e and e['metadata'].get('contentSubtype') == 'LIVE': r.append(e['metadata']) return list({item['contentId']: item for item in r}.values()) def get_single(token, content_id): headers = {'entitlementtoken': token, 'user-agent': user_agent, 'x-f1-device-info': 'device=web;screen=browser;os=windows;browser=chrome;browserVersion=139.0.0.0;osVersion=10;appVersion=release-R45.0.2;playerVersion=8.212.0'} params = {'contentId': content_id} response = req.get('https://f1tv.formula1.com/2.0/R/ENG/WEB_HLS/ALL/CONTENT/PLAY', params=params, headers=headers) data = response.json() result_obj = data.get('resultObj', {}) return result_obj.get('url', ''), result_obj.get('laURL', ''), result_obj.get('entitlementToken', '') def get_entitlement(ascendon_token): headers = {'ascendontoken': ascendon_token, 'user-agent': user_agent, 'x-f1-device-info': 'device=web;screen=browser;os=windows;browser=chrome;browserVersion=139.0.0.0;osVersion=10;appVersion=release-R45.0.2;playerVersion=8.212.0'} response = req.get('https://f1tv.formula1.com/2.0/R/ENG/WEB_DASH/ALL/USER/ENTITLEMENT', headers=headers) return response.json().get('resultObj', {}).get('entitlementToken', '') def login(): print("Please provide ascendon_token manually in the tokens file", file=sys.stderr) print("Login to F1TV website, go to DevTools -> Application -> Cookies and copy 'login-session' value", 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)) ascendon_token = auth['ascendonToken'] try: jwt.decode(ascendon_token, options={"verify_signature": False}) except: decoded = urllib.parse.unquote(ascendon_token) data = json.loads(decoded) ascendon_token = data['data']['subscriptionToken'] token = get_entitlement(ascendon_token) except: return "error" if action == "channels" or action == "events": output = {'Channels' if action == "channels" else 'Events': []} for c in get_events(): item = {'Name': c['title'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(c['contentId']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(c['contentId']), 'Video': 'best'} if action == "events": item['Autostart'] = True item['Start'] = int(datetime.datetime.now(pytz.UTC).timestamp()) item['End'] = int((datetime.datetime.now(pytz.UTC) + datetime.timedelta(hours=4)).timestamp()) output['Channels' if action == "channels" else 'Events'].append(item) print(json.dumps(output, indent=2)) elif action == "manifest": content_id = id.replace('id=', '') if id.startswith('id=') else id url, lic_url, lic_token = get_single(token, content_id) pssh_data = get_pssh_from_mpd(url) if url else None output = {"Cdn": [], "ManifestUrl": url, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "LicenseUrl": lic_url, "LicenseToken": lic_token} if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) elif action == "cdm" and cdm_param == "external": content_id = id.replace('id=', '') if id.startswith('id=') else id url, lic_url, lic_token = get_single(token, content_id) pssh_data = get_pssh_from_mpd(url) if url else None if pssh_data and lic_url: for key in do_cdm_external(pssh_data, lic_url, lic_token): print(key) if do_action() == "error": print("Error: Please create token file with ascendonToken", file=sys.stderr) sys.exit(1)