#!/usr/bin/python3 import sys import os import o11 import base64 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') token_param = o11.parse_params(sys.argv, 'token') 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 = o11.parse_params(sys.argv, 'cdm') drm = o11.parse_params(sys.argv, 'drm') kid = o11.parse_params(sys.argv, 'kid') pssh = o11.parse_params(sys.argv, 'pssh') challenge = o11.parse_params(sys.argv, 'challenge') heartbeaturl = o11.parse_params(sys.argv, 'heartbeaturl') heartbeatparams = o11.parse_params(sys.argv, 'heartbeatparams') o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" authFile = '/SFR_' + user + '.tokens' proxies = {"http": proxy, "https": proxy} if proxy else {} user_agent = 'GEN8 - 3.2.0 - samsung SM - mobile - Android OS' def do_cdm_external(pssh_data, device_id, token_sso): """Extract Widevine keys using pywidevine library""" 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 = { 'Content-Type': 'application/octet-stream', 'User-Agent': user_agent, 'CustomData': f'deviceId={device_id}&deviceType=MOBILE&osName=Android&description=Android&type=LIVEOTT&appId=android&tokenType=castoken&tokenSSO={token_sso}&prefetch=false', 'Host': 'ws-asgard.tv.sfr.net', } licence = req.post('https://ws-backendtv.sfr.fr/asgard-drm-widevine/public/licence', headers=lic_headers, data=challenge_data) try: licence.raise_for_status() cdm_obj.parse_license(session_id, licence.content) keys = [] for key in cdm_obj.get_keys(session_id): if key.type != 'SIGNING': keys.append(f"{key.kid.hex}:{key.key.hex()}") cdm_obj.close(session_id) return keys except Exception as e: print(f'License request failed: {e}', file=sys.stderr) return [] 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 init_to_pssh(init_url): headers = {'user-agent': user_agent} response = req.get(init_url, headers=headers) return to_pssh(response.content)[0] if to_pssh(response.content) else None def get_pssh_and_location(url): headers = {'User-Agent': user_agent} response = req.get(url, headers=headers, allow_redirects=False) location = response.headers.get('location', url) response = req.get(location, headers=headers) soup = BeautifulSoup(response.content, features="xml") init = soup.find('SegmentTemplate') if init and init.get('initialization'): bandwidth = soup.find('Representation')['bandwidth'] init_url = location[:location.rfind('/') + 1] + init['initialization'].replace('$Bandwidth$', bandwidth) return init_to_pssh(init_url), location return None, location def create_token(b64_up): headers = { 'Accept': 'application/json; charset=utf-8', 'Authorization': 'Basic ' + b64_up, 'secret': 'Basic U0ZSVFY4QW5kcm9pZFYxOnBvbGx1dGlvbjIwMjA=', 'Content-Type': 'application/json', 'User-Agent': 'GEN8/3.2.0 (com.sfr.android.gen8; build:3203000; Android OS 9) okhttp/4.10.0', } response = req.get('https://sso-client.sfr.fr/cas/services/rest/3.2/createToken.json?duration=86400', headers=headers) return response.json()['createToken']['token'] def get_device_id(token_sso): headers = {'User-Agent': user_agent} url = 'https://ws-asgard.tv.sfr.net/asgard-domain/public/listDevice?token=' + urllib.parse.quote_plus(token_sso) response = req.get(url, headers=headers) data = response.json() return data[0]['deviceId'] if data else None def get_streams(token_sso): headers = {'User-Agent': user_agent} params = {'app': 'gen8', 'device': 'android', 'token': token_sso} response = req.get('https://ws-sekai.tv.sfr.net/sekai-service-plan/public/v2/service-list', headers=headers, params=params) return response.json() def login(): print("logging in...", file=sys.stderr) un_up = f'{user}:{password}' b64_up = base64.b64encode(un_up.encode()).decode() token_sso = create_token(b64_up) device_id = get_device_id(token_sso) auth_data = {'b64_up': b64_up, 'deviceId': device_id} 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_sso = create_token(auth['b64_up']) device_id = auth['deviceId'] except: return "error" if action == "channels": output = {'Channels': []} streams = get_streams(token_sso) for s in streams: if s.get('streams'): for stream in s['streams']: if stream.get('drm') == 'WIDEVINE': channel = { 'Name': s['name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(s.get('id', '')) + '&url=' + stream['url'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(s.get('id', '')) + '&url=' + stream['url'], 'Video': 'best', 'OnDemand': True, 'SpeedUp': True } output['Channels'].append(channel) break print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} streams = get_streams(token_sso) for s in streams: if s.get('streams'): for stream in s['streams']: if stream.get('drm') == 'WIDEVINE': event = { 'Name': s['name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(s.get('id', '')) + '&url=' + stream['url'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(s.get('id', '')) + '&url=' + stream['url'], 'Video': 'best', 'Autostart': True, 'Start': int(datetime.datetime.now(pytz.UTC).timestamp()), 'End': int((datetime.datetime.now(pytz.UTC) + datetime.timedelta(hours=4)).timestamp()) } output['Events'].append(event) break print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": url = o11.parse_params(sys.argv, 'url') if not url and '&url=' in id: parts = id.split('&url=') url = parts[1] if len(parts) > 1 else '' try: pssh_data, location = get_pssh_and_location(url) output = { "Cdn": [], "ManifestUrl": location, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "Heartbeat": {"Url": '', "Params": '', "PeriodMs": 5*60*1000}, "TokenSSO": token_sso, "DeviceId": device_id } if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) except Exception as e: print(f"Error getting manifest: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "internal": tok = o11.parse_params(sys.argv, 'tokenSSO') or token_sso dev_id = o11.parse_params(sys.argv, 'deviceId') or device_id try: lic_headers = { 'Content-Type': 'application/octet-stream', 'CustomData': f'deviceId={dev_id}&deviceType=MOBILE&osName=Android&description=Android&type=LIVEOTT&appId=android&tokenType=castoken&tokenSSO={tok}&prefetch=false', } response = req.post('https://ws-backendtv.sfr.fr/asgard-drm-widevine/public/licence', headers=lic_headers, data=base64.b64decode(challenge)) response_b64 = str(base64.b64encode(response.content), 'ascii') if response_b64.startswith('CA'): print(response_b64) else: print(response.text, file=sys.stderr) return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "external": url = o11.parse_params(sys.argv, 'url') if not url and '&url=' in id: parts = id.split('&url=') url = parts[1] if len(parts) > 1 else '' try: pssh_data, location = get_pssh_and_location(url) if pssh_data: keys = do_cdm_external(pssh_data, device_id, token_sso) for key in keys: print(key) else: print("No PSSH found in manifest", file=sys.stderr) return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" else: print("invalid action: " + action, file=sys.stderr) if do_action() == "error": login() do_action()