#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import urllib.parse 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 = '/OrangeTV_' + user + '.tokens' user_agent = 'Dalvik/2.1.0 (Linux; U; Android 9; SM-G955F Build/PPR1.180610.011)' def do_cdm_external(pssh_data, cas_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 = {'Connection': 'close', 'Content-Type': 'application/octet-stream', 'User-Agent': user_agent, 'Host': 'android.orangetv.orange.es', 'Accept-Encoding': 'gzip'} licence = req.post('https://android.orangetv.orange.es/mob/api/rtv/v1/widevinedrm?token=' + cas_token, 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, res='high'): if res != 'low': last = url.split('/')[-1] mlast = last.replace('low', res).replace('medium', res) url = url.replace(last, mlast) headers = {'Accept': '*/*', 'Referer': 'https://orangetv.orange.es/', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36'} try: 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': return cp.find('cenc:pssh').text, response.url except: if res == 'high': return get_pssh_from_mpd(url, 'medium') elif res == 'medium': return get_pssh_from_mpd(url, 'low') return None, url def get_channels(): headers = {'Host': 'android.orangetv.orange.es', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'User-Agent': 'okhttp/4.10.0'} params = {'client': 'json', 'bouquet_external_id': '1_PRO', 'model_external_id': 'Smartphone_Android', 'filter_unsupported_channels': 'true'} response = req.get('https://android.orangetv.orange.es/mob/api/rtv/v1/GetChannelList', params=params, headers=headers) return response.json().get('response', []) def get_single(identity, serial_number, household_id, channel_id): headers = {'Host': 'android.orangetv.orange.es', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'Cookie': identity, 'User-Agent': 'okhttp/4.10.0', 'HouseholdID': household_id} params = {'client': 'json', 'serial_number': serial_number, 'include_cas_token': 'true', 'channel_external_id': channel_id} response = req.get('https://android.orangetv.orange.es/mob/api/rtv/v1/GetLivePlayingInfo', headers=headers, params=params) data = response.json().get('response', {}) return data.get('playingUrl', ''), urllib.parse.quote_plus(data.get('casToken', '')) def do_login_request(username, pwd): headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'android.orangetv.orange.es', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'User-Agent': 'okhttp/4.10.0'} data = f'client=json&username={username}&password={pwd}' response = req.post(f'https://android.orangetv.orange.es/mob/api/rtv/v1/Login?username={username}', headers=headers, data=data) return response.json().get('response', {}).get('message', '') def get_serial_number(identity): headers = {'Host': 'android.orangetv.orange.es', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'Cookie': identity, 'User-Agent': 'okhttp/4.10.0'} response = req.get('https://android.orangetv.orange.es/mob/api/rtv/v1/GetTerminalList?client=json', headers=headers) data = response.json().get('response', {}) terminals = data.get('terminals', []) for t in terminals: if t.get('model', {}).get('externalId') in ['PC', 'Smartphone_Android']: return str(data.get('householdId', '')), t.get('serialNumber', '') return '', '' def login(): print("logging in...", file=sys.stderr) identity = do_login_request(user, password) household_id, serial_number = get_serial_number(identity) auth_data = {'identity': identity, 'serial_number': serial_number, 'household_id': household_id} json.dump(auth_data, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) print("logged in successfully", file=sys.stderr) def get_auth(): auth = json.load(open(os.path.abspath(os.path.dirname(__file__)) + authFile)) return auth['identity'], auth['serial_number'], auth['household_id'] def do_action(): if action == "login": login() sys.exit() try: identity, serial_number, household_id = get_auth() except: return "error" if action == "channels": output = {'Channels': []} for c in get_channels(): output['Channels'].append({'Name': c['name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'eid=' + c['externalChannelId'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'eid=' + c['externalChannelId'], 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in get_channels(): output['Events'].append({'Name': c['name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'eid=' + c['externalChannelId'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'eid=' + c['externalChannelId'], '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('eid=', '') if id.startswith('eid=') else id url, cas_token = get_single(identity, serial_number, household_id, channel_id) pssh_data, final_url = get_pssh_from_mpd(url) output = {"Cdn": [], "ManifestUrl": final_url, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "CasToken": cas_token} if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) elif action == "cdm" and cdm_param == "external": channel_id = id.replace('eid=', '') if id.startswith('eid=') else id url, cas_token = get_single(identity, serial_number, household_id, channel_id) pssh_data, final_url = get_pssh_from_mpd(url) if pssh_data and cas_token: for key in do_cdm_external(pssh_data, cas_token): print(key) if do_action() == "error": login() do_action()