#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import secrets import string 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 = '/FuboTV_' + user + '.tokens' user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36' def generate_random_string(length): alphabet = string.ascii_letters + string.digits return ''.join(secrets.choice(alphabet) for _ in range(length)) def do_cdm_external(pssh_data): 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': '*/*', 'cache-control': 'no-cache', 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'content-type': 'application/x-www-form-urlencoded'} licence = req.post('https://irdeto.fubo.tv/licenseServer/widevine/v1/FuboTV/license', 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': '*/*', 'cache-control': 'no-cache', 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'user-agent': user_agent} response = req.get(url, headers=headers) for cp in BeautifulSoup(response.content, features='xml').findAll('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 return None def get_channels(token): headers = {'accept': '*/*', 'authorization': 'Bearer ' + token, 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'user-agent': user_agent} response = req.get('https://api.fubo.tv/epg', headers=headers) return response.json().get('response', []) def get_single(token, channel_id): headers = {'accept': '*/*', 'authorization': 'Bearer ' + token, 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'user-agent': user_agent, 'x-supported-streaming-protocols': 'hls,dash'} response = req.get(f'https://api.fubo.tv/v3/kgraph/v3/networks/{channel_id}/stream', headers=headers) return response.json().get('streamUrls', [{}])[0].get('url', '') def do_login_request(username, pwd): headers = {'accept': '*/*', 'content-type': 'application/json', 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'user-agent': user_agent, 'x-ad-id': generate_random_string(20), 'x-application-id': 'fubo', 'x-device-id': generate_random_string(20), 'x-device-app': 'web', 'x-device-platform': 'desktop'} json_data = {'email': username, 'password': pwd} response = req.put('https://api.fubo.tv/signin', headers=headers, json=json_data) data = response.json() return data.get('access_token', ''), data.get('refresh_token', '') def do_refresh(refresh_token): headers = {'accept': '*/*', 'authorization': 'Bearer ' + refresh_token, 'origin': 'https://www.fubo.tv', 'referer': 'https://www.fubo.tv/', 'user-agent': user_agent} response = req.post('https://api.fubo.tv/refresh', headers=headers, data='') data = response.json() return data.get('access_token', ''), data.get('refresh_token', '') def login(): print("logging in...", file=sys.stderr) token, refresh_token = do_login_request(user, password) auth_data = {'refresh_token': refresh_token} json.dump(auth_data, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) print("logged in 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)) token, refresh_token = do_refresh(auth['refresh_token']) auth['refresh_token'] = refresh_token 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): ch = c.get('data', {}).get('channel', {}) output['Channels'].append({'Name': ch.get('name', ''), 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + ch.get('id', ''), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + ch.get('id', ''), 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in get_channels(token): ch = c.get('data', {}).get('channel', {}) output['Events'].append({'Name': ch.get('name', ''), 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + ch.get('id', ''), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + ch.get('id', ''), '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 = get_single(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}}} 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 = get_single(token, channel_id) pssh_data = get_pssh_from_mpd(url) if pssh_data: for key in do_cdm_external(pssh_data): print(key) if do_action() == "error": login() do_action()