#!/usr/bin/python3 import sys import os import base64 import json import datetime import pytz import jwt import hmac import hashlib import time import string import random import secrets from urllib import parse # Add parent directory to path for o11 import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import o11 from bs4 import BeautifulSoup from pywidevine.cdm import Cdm from pywidevine.device import Device from pywidevine.pssh import PSSH # Parse command line parameters user = o11.parse_params(sys.argv, 'user') password = o11.parse_params(sys.argv, 'password') device = o11.parse_params(sys.argv, 'device') pin = o11.parse_params(sys.argv, 'pin') 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') # Session setup o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" # Configuration WVD_PATH = './WVD.wvd' authFile = '/MovistarCL_auth.json' SCRIPT_DIR = os.path.abspath(os.path.dirname(__file__)) USER_AGENT = 'cl.movistarplay/25.5.100 (Linux;Android 9) ExoPlayerLib/2.18.7 MPLAY/308/25.05.100' CONSUMER_KEY = 'stv_androidtv_2505100_cl' CONSUMER_SECRET = 'afa7Cm1eaBqSIh9LA7rxGEtWxjS3RMVM' DEVICE_FUT = '3080001' PROFILE_ID = '0' DEVICE_TYPE = '308' INSTANCE_ID = '26' SIGNATURE_METHOD = 'HMAC-SHA1' SIGNATURE_VERSION = '1.0' bearer_token = '' def get_auth(): try: return json.load(open(SCRIPT_DIR + authFile)) except: return None def save_auth(auth_data): json.dump(auth_data, open(SCRIPT_DIR + authFile, 'w'), indent=2) def generate_random_string(length=11): characters = string.ascii_letters + string.digits return ''.join(random.choice(characters) for _ in range(length)) def hmac_sha1(input_str, key): raw = input_str.encode("utf-8") key = key.encode('utf-8') hashed = hmac.new(key, raw, hashlib.sha1) return base64.encodebytes(hashed.digest()).decode('utf-8').strip() def get_signature(method, url, token, params, data): curr_time = round(time.time()) hmac_params = {'consumer_key': CONSUMER_KEY, 'nonce': base64.b64encode((generate_random_string() + str(curr_time)).encode('utf-8')).decode('utf-8'), 'signature_method': SIGNATURE_METHOD, 'timestamp': curr_time, 'version': SIGNATURE_VERSION} endpoint = url.split('?')[0].replace('https://', '') input_json = {} input_json.update(params) input_json.update(data) input_json.update(hmac_params) sorted_dict = {key: value for key, value in sorted(input_json.items())} input_str = '' for key, value in sorted_dict.items(): if key == 'userName' or key == 'password': value = parse.quote(value) input_str += f'&{key}={value}' input_str = f'{method}&{parse.quote_plus(endpoint)}&{parse.quote_plus(token)}' + parse.quote_plus(input_str) signature = hmac_sha1(input_str, CONSUMER_SECRET) return base64.b64encode(f'consumer_key="{hmac_params["consumer_key"]}",nonce="{hmac_params["nonce"]}",signature_method="{hmac_params["signature_method"]}",signature="{signature}",timestamp={hmac_params["timestamp"]},version="{hmac_params["version"]}"'.encode('utf-8')).decode('utf8') def do_refresh(device_id): url = 'https://canoway-cl.cdn.telefonica.com/instance/26/service/nsh/home' headers = {'User-Agent': USER_AGENT, 'Accept': 'application/json, text/javascript', 'Content-Type': 'application/x-www-form-urlencoded', 'X-Requested-With': 'cl.movistarplay'} token = f'Basic=' + base64.b64encode(f'{PROFILE_ID}|{device_id}:{DEVICE_TYPE},{DEVICE_FUT}'.encode('utf-8')).decode('utf-8') signature = get_signature('GET', url, token, params={}, data={}) headers['Authorization'] = f'OP {token}, Signature={signature}' response = req.get(url, headers=headers) response.raise_for_status() token = response.headers['x-op-access-token'] decoded = jwt.decode(token, options={'verify_signature': False}) if not decoded['device']: raise Exception('No device') return token def login(): global bearer_token print("logging in...", file=sys.stderr) auth = get_auth() if auth and 'device_id' in auth: try: bearer_token = do_refresh(auth['device_id']) print("logged in successfully", file=sys.stderr) return bearer_token except: pass if not auth or 'device_id' not in auth: print("No auth found. Please add MovistarCL device_id to auth file.", file=sys.stderr) print("Login and register device first", file=sys.stderr) save_auth({'device_id': ''}) sys.exit(1) print("Login failed", file=sys.stderr) sys.exit(1) def get_token(): global bearer_token if bearer_token: return bearer_token return login() def get_pssh_from_mpd(url): response = req.get(url, headers={'user-agent': USER_AGENT}) content_protections = BeautifulSoup(response.content, features="xml").findAll('ContentProtection') for cp in content_protections: if cp.get('schemeIdUri', '').lower() == 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed': pssh_elem = cp.find('cenc:pssh') if pssh_elem: return pssh_elem.text return None def get_playback_id(): decoded = jwt.decode(bearer_token, options={'verify_signature': False}) url = 'https://ucand.movistarplay.cl/service3.0/UserService.svc/GetUser' headers = {'User-Agent': USER_AGENT, 'Accept': 'application/json, text/javascript', 'Content-Type': 'application/x-www-form-urlencoded', 'X-Requested-With': 'cl.movistarplay'} params = {'token': decoded['ut'], 'oauth_consumer_key': CONSUMER_KEY, 'oauth_nonce': generate_random_string(), 'oauth_signature_method': SIGNATURE_METHOD, 'oauth_timestamp': round(time.time()), 'oauth_version': SIGNATURE_VERSION} response = req.get(url, params=params, headers=headers) try: return response.json()['Content']['PlaybackId'] except: return None def do_cdm_internal(challenge_b64): decoded = jwt.decode(bearer_token, options={'verify_signature': False}) lic_headers = {'Accept': '*/*', 'User-Agent': USER_AGENT, 'dt-custom-data': decoded['ut'], 'x-authentication': decoded['ut'], 'Content-Type': 'application/octet-stream'} response = req.post('https://janus-cl.cdn.telefonica.com/DashIssuer.aspx', headers=lic_headers, data=base64.b64decode(challenge_b64)) response_b64 = str(base64.b64encode(response.content), 'ascii') if response_b64.startswith('CA'): return response_b64 return None def do_cdm_external(pssh_b64): try: pssh_obj = PSSH(pssh_b64) 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) decoded = jwt.decode(bearer_token, options={'verify_signature': False}) lic_headers = {'Accept': '*/*', 'User-Agent': USER_AGENT, 'dt-custom-data': decoded['ut'], 'x-authentication': decoded['ut'], 'Content-Type': 'application/octet-stream'} licence = req.post('https://janus-cl.cdn.telefonica.com/DashIssuer.aspx', 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 except Exception as e: print(f'CDM external failed: {e}', file=sys.stderr) return None def do_action(): get_token() if action == "login": login() sys.exit() if action == "channels": output = {'Channels': []} url = 'https://canoway-cl.cdn.telefonica.com/instance/26/service/nsh/home' headers = {'User-Agent': USER_AGENT, 'Accept': 'application/json, text/javascript', 'Content-Type': 'application/x-www-form-urlencoded', 'X-Requested-With': 'cl.movistarplay'} token = f'Bearer={bearer_token}' signature = get_signature('GET', url, token, params={}, data={}) headers['Authorization'] = f'OP {token}, Signature={signature}' response = req.get(url, headers=headers) try: data = response.json() decoded = jwt.decode(bearer_token, options={'verify_signature': False}) account_subscriptions = decoded['subscriptions'].split('|') for ch in data.get('TvChannels', []): for a_s in account_subscriptions: if a_s in ch.get('SubscriptionPids', []): channel = { 'Name': ch.get('CallLetter', 'Unknown'), 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={ch.get('ChannelId', '')}", 'CdmType': 'widevine', 'UseCdm': True, 'Cdm': f"id={ch.get('ChannelId', '')}", 'Video': 'best', 'OnDemand': True, 'SpeedUp': True, } output['Channels'].append(channel) break print(json.dumps(output, indent=2)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "events": output = {'Events': []} print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": try: channel_id = id playback_id = get_playback_id() video_url = f"https://example.com/stream/{channel_id}.mpd" if playback_id: video_url = video_url.replace('_-_psi_-_', playback_id) output = { "Cdn": [{"Name": "default", "ManifestUrl": video_url}], "ManifestUrl": video_url, "Headers": {"Manifest": {'User-Agent': USER_AGENT}, "Media": {'User-Agent': USER_AGENT}}, "Heartbeat": {"Url": '', "Params": '', "PeriodMs": 5*60*1000} } print(json.dumps(output)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "internal": try: result = do_cdm_internal(challenge) if result: print(result) else: return "error" except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "external": try: pssh_to_use = pssh if pssh_to_use: keys = do_cdm_external(pssh_to_use) if keys: for key in keys: print(key) else: return "error" else: 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()