#!/usr/bin/env python3 """ TIM Vision - Xaccel Manager con WVD locale Flow: createToken -> InitLiveTV -> WSXUrl -> MPD -> PSSH -> GetLicence -> keys """ import os, sys, re, json, time, base64, threading, uuid import xml.etree.ElementTree as ET from datetime import datetime from typing import Dict, List, Optional, Tuple from urllib.parse import urlparse, parse_qs, urlencode, urlunparse import requests from pywidevine import PSSH, Cdm, Device GREEN='\033[32m'; YELLOW='\033[33m'; RED='\033[31m'; CYAN='\033[36m'; BLUE='\033[34m'; RESET='\033[0m' def print_banner(): print(f"""{YELLOW} _____ _ __ ___ _ |_ _(_)_ __ __\ \ / (_)___(_) ___ _ __ | | | | '_ ` _ \ \ / /| / __| |/ _ \| '_ \ | | | | | | | | \ V / | \__ \ | (_) | | | | |_| |_|_| |_| |_|\_/ |_|___/_|\___/|_| |_| {BLUE}{'─' * 70}{RESET} {CYAN} ◆ TIM VISION · Xaccel Manager betaV1.2 · by AntheliaTV ◆{RESET} {BLUE}{'─' * 70}{RESET} """) CONFIG_FILE = "timvision_config.json" CACHE_FILE = "pssh_cache.json" WVD_FILE = "WVD.wvd" TOKEN_FILE = "timvision_tokens.json" CHANNELS = [ {"epgid":"90099000","name":"TIMVISION", "wsx":"timvisionpromo1"}, {"epgid":"90200132","name":"Rete 4", "wsx":"rete4"}, {"epgid":"90200130","name":"Canale 5", "wsx":"canale5"}, {"epgid":"90200131","name":"Italia 1", "wsx":"italia1"}, {"epgid":"90300133","name":"LA7 HD", "wsx":"la7"}, {"epgid":"90320074","name":"NOVE", "wsx":"novehd"}, {"epgid":"90100897","name":"20 Mediaset", "wsx":"venti"}, {"epgid":"90100475","name":"Iris", "wsx":"iris"}, {"epgid":"90120081","name":"27 TwentySeven", "wsx":"twentyseven"}, {"epgid":"90300553","name":"LA7 Cinema HD", "wsx":"la7dhd"}, {"epgid":"90100567","name":"La5", "wsx":"la5"}, {"epgid":"90020075","name":"Real Time", "wsx":"realtimehd"}, {"epgid":"90020062","name":"Food Network", "wsx":"foodnetworkhd"}, {"epgid":"90120088","name":"Cine34", "wsx":"cine34"}, {"epgid":"90100896","name":"Focus", "wsx":"focus"}, {"epgid":"90020037","name":"Discovery", "wsx":"warnertv"}, {"epgid":"90120080","name":"Giallo", "wsx":"giallo"}, {"epgid":"90100721","name":"TOPcrime", "wsx":"topcrime"}, {"epgid":"90120077","name":"K2", "wsx":"k2"}, {"epgid":"90120078","name":"Frisbee", "wsx":"frisbee"}, {"epgid":"90100895","name":"Tgcom 24", "wsx":"tgcom24"}, {"epgid":"90020076","name":"DMAX", "wsx":"dmaxhd"}, {"epgid":"90100603","name":"Mediaset Extra", "wsx":"mdsextra"}, {"epgid":"90020061","name":"HGTV", "wsx":"hgtvhd"}, {"epgid":"90020060","name":"Motor Trend", "wsx":"motortrendhd"}, {"epgid":"90200642","name":"Italia 2", "wsx":"italia2"}, {"epgid":"90020020","name":"CNN International", "wsx":"cnni"}, {"epgid":"90020002","name":"Eurosport 3", "wsx":"eurosport3"}, {"epgid":"90020003","name":"Eurosport 4", "wsx":"eurosport4"}, {"epgid":"90020004","name":"Eurosport 5", "wsx":"eurosport5"}, {"epgid":"90020005","name":"Eurosport 6", "wsx":"eurosport6"}, {"epgid":"90020000","name":"Eurosport 1", "wsx":"eurosport1"}, {"epgid":"90020001","name":"Eurosport 2", "wsx":"eurosport2"}, {"epgid":"90020021","name":"Euronews", "wsx":"euronews"}, ] pssh_cache = {} monitor_running = False monitor_thread = None # ── Config ───────────────────────────────────────────────────────────────────── def load_config(): if os.path.exists(CONFIG_FILE): return json.load(open(CONFIG_FILE)) cfg = { "xaccel": {"base_url":"http://117.55.203.19","token":"cEdvDHHvdw7K1f7S2xUPKqiQ"}, "timvision": { "proxy": { "http":"http://qC393GkhUwyLy1k:IH7ob2O2MSJ0FSw@82.41.254.132:49938", "https":"http://qC393GkhUwyLy1k:IH7ob2O2MSJ0FSw@82.41.254.132:49938" }, "portail_id": "vbdTj7eb6aM.", "pass_id": "", }, "monitor": {"check_interval":60,"restart_delay":10} } json.dump(cfg, open(CONFIG_FILE,'w'), indent=2) return cfg def load_cache(): global pssh_cache if os.path.exists(CACHE_FILE): try: pssh_cache = json.load(open(CACHE_FILE)) except: pssh_cache = {} def save_cache(): json.dump(pssh_cache, open(CACHE_FILE,'w'), indent=2) def log(msg, color=RESET): print(f"{color}[{datetime.now().strftime('%H:%M:%S')}] {msg}{RESET}", flush=True) # ── Auth flow ────────────────────────────────────────────────────────────────── UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" def create_token(config, pass_id=None): """POST createToken -> passToken + passId""" portail_id = config['timvision'].get('portail_id','vbdTj7eb6aM.') if not pass_id: pass_id = config['timvision'].get('pass_id','') data = { 'portailId': portail_id, 'media': 'web', 'vect': 'INTERNET', 'passIdType':'pass', 'noCache': 'false', } if pass_id: data['passId'] = pass_id h = { 'User-Agent': UA, 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'Content-Type': 'application/x-www-form-urlencoded', } r = requests.post( 'https://auth-api-provider-cdn.prod.pass.tim.cptech.pro/provider/services/tiita-it/public/createToken', headers=h, data=data, proxies=config['timvision']['proxy'], timeout=20 ) r.raise_for_status() resp = r.json()['response'] return resp['passToken'], resp.get('passId','') def init_live_tv(config, pass_token): """POST InitLiveTV -> LiveToken + lista canali""" device_key_id = '1784042923370-bd0611cf72e6' user_key_id = '79426aed-1fa3-4bb7-9a01-5c6940fdf753' payload = { "ServiceRequest": { "InData": { "UseRmuTokenPlaceHolder": True, "PassData": {"Id": 0, "Token": pass_token}, "UserKeyId": user_key_id, "DeviceKeyId": device_key_id, "PDSData": {"GroupTypes": "1;4"}, "DeviceSpecificities": {"BpkCompatibility": True} } } } h = { 'User-Agent': UA, 'Content-Type': 'application/json', 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'Accept': 'application/json, text/plain, */*', } r = requests.post( 'https://dtd-pfsbo-slc-ltv-api-eu-west-1.prod.bo.tim.cptech.pro/api/V4/zones/tiita/locations/it/devices/3/apps/1/jobs/InitLiveTV', json=payload, headers=h, proxies=config['timvision']['proxy'], timeout=30 ) r.raise_for_status() out = r.json()['ServiceResponse']['OutData'] live_token = out['LiveToken'] device_key_id_ret = device_key_id return live_token, user_key_id, device_key_id_ret, out def do_auth(config): """Esegue il flow completo e salva i token""" log('Auth TIM Vision...', CYAN) pass_token, pass_id = create_token(config) log(f'passToken OK', GREEN) live_token, user_key_id, device_key_id, _ = init_live_tv(config, pass_token) log(f'LiveToken OK', GREEN) tokens = { 'live_token': live_token, 'pass_token': pass_token, 'pass_id': pass_id, 'user_key_id': user_key_id, 'device_key_id': device_key_id, 'timestamp': time.time(), } json.dump(tokens, open(TOKEN_FILE,'w'), indent=2) # aggiorna config config['timvision']['live_token'] = live_token config['timvision']['pass_id'] = pass_id config['timvision']['user_key_id'] = user_key_id config['timvision']['device_key_id'] = device_key_id json.dump(config, open(CONFIG_FILE,'w'), indent=2) return tokens def get_tokens(config): """Carica token salvati o fa auth""" if os.path.exists(TOKEN_FILE): t = json.load(open(TOKEN_FILE)) age = time.time() - t.get('timestamp', 0) if age < 3600 * 8: # validi 8h return t return do_auth(config) # ── Stream ───────────────────────────────────────────────────────────────────── def get_mpd_url(wsx_url, config): h = { 'User-Agent': UA, 'Authorization': f"Bearer {config['timvision']['live_token']}", 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'xx-operator': 'pc', 'xx-service': 'mycanal', } r = requests.get(wsx_url, headers=h, proxies=config['timvision']['proxy'], timeout=15, allow_redirects=True) if not r.ok: return None if '.mpd' in r.url: return r.url try: data = r.json() def find_url(obj): if isinstance(obj, dict): for k,v in obj.items(): if k in ('src','url','href') and isinstance(v,str) and v.startswith('http'): return v res = find_url(v) if res: return res elif isinstance(obj, list): for i in obj: res = find_url(i) if res: return res return find_url(data) except: return None def get_manifest(mpd_url, config): h = { 'User-Agent': UA, 'Authorization': f"Bearer {config['timvision']['live_token']}", } r = requests.get(mpd_url, headers=h, proxies=config['timvision']['proxy'], timeout=30) return r.text if r.ok else None def extract_pssh(mpd_content): pssh_list = [] try: root = ET.fromstring(mpd_content) for cp in root.findall('.//{urn:mpeg:dash:schema:mpd:2011}ContentProtection'): for pe in cp.findall('{urn:mpeg:cenc:2013}pssh'): if pe.text and pe.text.strip() not in pssh_list: pssh_list.append(pe.text.strip()) except: pass return pssh_list # ── License con WVD locale ───────────────────────────────────────────────────── def get_keys_wvd(pssh_b64, epgid, config, tokens): global pssh_cache cache_key = f"{epgid}:{pssh_b64[:20]}" if cache_key in pssh_cache: cached = pssh_cache[cache_key] log(f'Cache hit: {cached}', GREEN) return cached.get('kid'), cached.get('key') try: device = Device.load(WVD_FILE) cdm = Cdm.from_device(device) session = cdm.open() challenge = cdm.get_license_challenge(session, PSSH(pssh_b64)) challenge_b64 = base64.b64encode(challenge).decode() payload = { "ServiceRequest": { "InData": { "EpgId": int(epgid), "LiveToken": tokens['live_token'], "UserKeyId": tokens['user_key_id'], "DeviceKeyId": tokens['device_key_id'], "ChallengeInfo": challenge_b64, "Mode": "MKPL", "PersistentLicense": "false", } } } h = { 'User-Agent': UA, 'Content-Type': 'application/json', 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', } r = requests.post( 'https://dtd-pfsbo-slc-ltv-api-eu-west-1.prod.bo.tim.cptech.pro/api/V4/zones/tiita/devices/31/apps/1/jobs/GetLicence', json=payload, headers=h, proxies=config['timvision']['proxy'], timeout=30 ) if not r.ok: log(f'GetLicence {r.status_code}: {r.text[:200]}', RED) cdm.close(session) return None, None resp = r.json() license_b64 = resp['ServiceResponse']['OutData']['LicenseInfo'] license_bytes = base64.b64decode(license_b64) cdm.parse_license(session, license_bytes) kid = key = None for k in cdm.get_keys(session): if k.type != 'SIGNING': kid = k.kid.hex key = k.key.hex() break cdm.close(session) if kid and key: pssh_cache[cache_key] = {'kid': kid, 'key': key} save_cache() log(f'Keys OK — KID:{kid[:16]}... KEY:{key[:16]}...', GREEN) return kid, key except Exception as e: log(f'WVD error: {e}', RED) return None, None def get_channel_keys(channel, config, tokens): wsx_url = f"https://dtd-pfsbo-ottd-rmu-front-eu-west-1.prod.bo.tim.cptech.pro/plfiles/v2/ital/dash-ssl/{channel['wsx']}.json?device=PC&capabilities=0x4" mpd_url = get_mpd_url(wsx_url, config) if not mpd_url: log(f'{channel["name"]}: MPD non disponibile', RED) return None, None, None mpd = get_manifest(mpd_url, config) if not mpd: log(f'{channel["name"]}: manifest non disponibile', RED) return mpd_url, None, None pssh_list = extract_pssh(mpd) if not pssh_list: log(f'{channel["name"]}: PSSH non trovato', RED) return mpd_url, None, None # Usa il secondo PSSH (Widevine) pssh_b64 = pssh_list[1] if len(pssh_list) > 1 else pssh_list[0] kid, key = get_keys_wvd(pssh_b64, channel['epgid'], config, tokens) return mpd_url, kid, key # ── Xaccel ──────────────────────────────────────────────────────────────────── def xaccel_get_streams(config): try: r = requests.get(f"{config['xaccel']['base_url']}/api/streams", params={"token": config['xaccel']['token']}, timeout=10) if r.ok: data = r.json() streams = data if isinstance(data,list) else data.get('streams',data.get('data',[])) return {s.get('name'):s for s in streams if s and s.get('name')} except: pass return {} def xaccel_send(name, url, epgid, kid, key, stream_id=None, config=None): payload = { "token": config['xaccel']['token'], "url": url, "name": name, "epgid": epgid, "drm": {"type":"widevine","keys":{kid:key}} } if stream_id: payload["id"] = stream_id ep = f"{config['xaccel']['base_url']}/api/stream/{'update' if stream_id else 'add'}" try: r = requests.post(ep, json=payload, timeout=10) return r.ok except: return False def build_url(mpd_url, kid, key): if not kid or not key: return mpd_url p = urlparse(mpd_url) q = parse_qs(p.query) # Rimuovi eventuali chiavi precedenti q.pop('decryption_key', None) q.pop('v', None) # Ricostruisci query base (senza decryption_key) base_query = urlencode(q, doseq=True) # Aggiungi decryption_key con ":" letterale (no %3A) dk = f"decryption_key={kid}:{key}&v=best" final_query = f"{base_query}&{dk}" if base_query else dk return urlunparse((p.scheme, p.netloc, p.path, p.params, final_query, p.fragment)) # ── Comandi ─────────────────────────────────────────────────────────────────── def cmd_send_channel(): config = load_config(); load_cache() tokens = get_tokens(config) print(f"\n{CYAN}📺 Canali disponibili:{RESET}") for i,ch in enumerate(CHANNELS,1): print(f" {i:2d}. {ch['name']:<20} ({ch['epgid']})") choice = input(f"\n{CYAN}Scegli canale: {RESET}").strip() if not choice.isdigit(): return idx = int(choice)-1 if not 0 <= idx < len(CHANNELS): return ch = CHANNELS[idx] log(f'Canale: {ch["name"]}', YELLOW) mpd_url, kid, key = get_channel_keys(ch, config, tokens) if not mpd_url or not kid: return url = build_url(mpd_url, kid, key) existing = xaccel_get_streams(config) sid = existing.get(ch['name'],{}).get('id') action = 'AGGIORNAMENTO' if sid else 'CREAZIONE' print(f"\n{BLUE}URL: {url[:100]}...{RESET}") if input(f"\n{CYAN}Procedere con {action}? (s/n): {RESET}").strip().lower() != 's': return if xaccel_send(ch['name'], url, ch['epgid'], kid, key, sid, config): log(f'{action} OK!', GREEN) else: log(f'Errore {action.lower()}', RED) def cmd_extract_all(): config = load_config(); load_cache() tokens = get_tokens(config) results = []; found = 0 for i,ch in enumerate(CHANNELS,1): print(f"\n[{i}/{len(CHANNELS)}] {ch['name']}...", end=' ', flush=True) mpd_url, kid, key = get_channel_keys(ch, config, tokens) if kid and key: results.append({**ch, 'manifest':mpd_url, 'kid':kid, 'key':key, 'xaccel_url':build_url(mpd_url,kid,key)}) print(f"{GREEN}✅ {kid}:{key[:16]}...{RESET}") found += 1 else: results.append({**ch,'manifest':mpd_url,'kid':None,'key':None,'error':'non trovato'}) print(f"{RED}❌{RESET}") out = f"channels_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" json.dump({'timestamp':datetime.now().isoformat(),'total':len(CHANNELS),'found':found,'channels':results}, open(out,'w'), indent=2) print(f"\n{GREEN}✅ {found}/{len(CHANNELS)} — {out}{RESET}") def monitor_loop(): global monitor_running config = load_config(); load_cache() log('Monitor avviato', CYAN) while monitor_running: try: tokens = get_tokens(config) streams = xaccel_get_streams(config) if not streams: time.sleep(60); continue log(f'Controllo {len(streams)} canali...', CYAN) for name,data in streams.items(): sid = data.get('id'); url = data.get('url','') try: ok = requests.get(url, timeout=10, headers={'User-Agent':UA}).ok except: ok = False if ok: print(f" 🟢 {name}") else: print(f" 🔴 {name}: riavvio...") ch = next((c for c in CHANNELS if c['name']==name),None) if not ch: continue mpd_url,kid,key = get_channel_keys(ch,config,tokens) if not kid: continue time.sleep(config['monitor'].get('restart_delay',10)) new_url = build_url(mpd_url,kid,key) if xaccel_send(name,new_url,ch['epgid'],kid,key,sid,config): print(f" ✅ {name}: OK") iv = config['monitor'].get('check_interval',60) for _ in range(iv): if not monitor_running: break time.sleep(1) except KeyboardInterrupt: break except Exception as e: log(f'Errore: {e}',RED); time.sleep(30) log('Monitor fermato', YELLOW) def cmd_monitor(): global monitor_running, monitor_thread if monitor_running: monitor_running = False if monitor_thread: monitor_thread.join(timeout=5) log('Monitor fermato', GREEN) else: monitor_running = True monitor_thread = threading.Thread(target=monitor_loop, daemon=True) monitor_thread.start() log('Monitor avviato', GREEN) def cmd_refresh_token(): config = load_config() tokens = do_auth(config) log(f'Token rinnovato OK', GREEN) def main(): print_banner() config = load_config() while True: print(f"\n{'='*50}") print(f"{CYAN}🎬 TIM VISION - XACCEL MANAGER (WVD){RESET}") print(f"Monitor: {GREEN+'ATTIVO' if monitor_running else RED+'FERMO'}{RESET}") print(f"{'='*50}") print(f" {GREEN}1.{RESET} Invia canale a Xaccel") print(f" {GREEN}2.{RESET} Monitor (Avvia/Ferma)") print(f" {GREEN}3.{RESET} Estrai tutti in JSON") print(f" {GREEN}4.{RESET} Rinnova token") print(f" {RED}0.{RESET} Esci") print(f"{'='*50}") c = input(f"{CYAN}Scegli: {RESET}").strip() if c=='0': break elif c=='1': cmd_send_channel() elif c=='2': cmd_monitor() elif c=='3': cmd_extract_all() elif c=='4': cmd_refresh_token() if __name__ == '__main__': main()