#!/usr/bin/python3 import sys import os import o11 import base64 import json import datetime import pytz import requests 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') device_param = 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') o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" authFile = '/VectraTV_' + user + '.tokens' proxies = {"http": proxy, "https": proxy} if proxy else {} user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36' def do_cdm_external(pssh_data, licence_url, session_id_val, video_id): """Extract Widevine keys using pywidevine library""" pssh_obj = PSSH(pssh_data) device_obj = Device.load(WVD_PATH) cdm_obj = Cdm.from_device(device_obj) cdm_session_id = cdm_obj.open() challenge_data = cdm_obj.get_license_challenge(cdm_session_id, pssh_obj) lic_headers = { 'authority': 'api.tvsmart.pl', 'accept': '*/*', 'origin': 'https://tvsmart.vectra.pl', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, 'content-type': 'application/x-www-form-urlencoded', } params = {'videoAssetId': video_id, 'videoSessionId': session_id_val} licence = req.post(licence_url, params=params, headers=lic_headers, data=challenge_data) try: licence.raise_for_status() cdm_obj.parse_license(cdm_session_id, licence.content) keys = [] for key in cdm_obj.get_keys(cdm_session_id): if key.type != 'SIGNING': keys.append(f"{key.kid.hex}:{key.key.hex()}") cdm_obj.close(cdm_session_id) return keys except Exception as e: print(f'License request failed: {e}', file=sys.stderr) return [] def get_pssh_from_mpd(url): headers = { 'accept': '*/*', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, } response = req.get(url, headers=headers) 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': if cp.find('cenc:pssh'): return cp.find('cenc:pssh').text, None # Fallback to PlayReady KID for cp in content_protections: if cp.get('schemeIdUri', '').lower() == 'urn:uuid:9a04f079-9840-4286-ab92-e65be0885f95': return cp.find('mspr:pro').text if cp.find('mspr:pro') else None, cp.get('default_KID', '').replace('-', '').lower() return None, None def get_video_session(token, channel_uuid): headers = { 'authority': 'api.tvsmart.pl', 'accept': 'application/json', 'access-control-allow-origin': '*', 'api-device': 'Chrome; 119; Windows; 10; Windows; 10;', 'authorization': 'Bearer ' + token, 'cache-control': 'no-cache', 'origin': 'https://tvsmart.vectra.pl', 'pragma': 'no-cache', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, } params = {'type': 'channel', 'platform': 'BROWSER', 'system': 'tvonline', 'language': 'pl'} response = req.get(f'https://api.tvsmart.pl/player/product/{channel_uuid}/configuration', params=params, headers=headers) data = response.json() return data['videoSession']['videoSessionId'], str(data['videoId']) def get_single(session_id_val, channel_uuid): headers = { 'authority': 'api.tvsmart.pl', 'accept': '*/*', 'cache-control': 'no-cache', 'origin': 'https://tvsmart.vectra.pl', 'pragma': 'no-cache', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, } params = {'type': 'channel', 'videoSessionId': session_id_val} response = req.get(f'https://api.tvsmart.pl/player/product/{channel_uuid}/playlist', params=params, headers=headers) data = response.json() sources = data['sources'] dash = sources['DASH'][0]['src'] if dash.startswith('//'): dash = 'https:' + dash return dash, data['drm']['WIDEVINE'] def get_channels(token): headers = { 'authority': 'api.tvsmart.pl', 'accept': 'application/json', 'access-control-allow-origin': '*', 'authorization': 'Bearer ' + token, 'origin': 'https://tvsmart.vectra.pl', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, } params = {'offset': '0', 'limit': '300', 'platform': 'BROWSER', 'system': 'tvonline', 'language': 'pl'} response = req.get('https://api.tvsmart.pl/products/channel', params=params, headers=headers) data = response.json() channels = data['data'] # Remove duplicates seen = set() unique = [] for c in channels: if c['title'] not in seen: seen.add(c['title']) unique.append(c) return unique def delete_session(token, session_id_val): headers = { 'authority': 'api.tvsmart.pl', 'accept': 'application/json', 'authorization': 'Bearer ' + token, 'origin': 'https://tvsmart.vectra.pl', 'referer': 'https://tvsmart.vectra.pl/', 'user-agent': user_agent, } params = {'platform': 'BROWSER', 'system': 'tvonline', 'language': 'pl'} req.delete(f'https://api.tvsmart.pl/player/videosession/{session_id_val}', params=params, headers=headers) def login(): print("Saving token...", file=sys.stderr) if not token_param: print("Please provide token parameter from browser DevTools -> Application -> Local Storage", file=sys.stderr) sys.exit(1) auth_data = {'token': token_param} json.dump(auth_data, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) print("Token saved 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 = auth['token'] except: return "error" if action == "channels": output = {'Channels': []} channels = get_channels(token) for chan in channels: channel = { 'Name': chan['title'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(chan['uuid']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['uuid']), 'Video': 'best', 'OnDemand': True, 'SpeedUp': True } output['Channels'].append(channel) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} channels = get_channels(token) for chan in channels: event = { 'Name': chan['title'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + str(chan['uuid']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + str(chan['uuid']), '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) print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": channel_uuid = id try: session_id_val, video_id = get_video_session(token, channel_uuid) dash_url, licence_url = get_single(session_id_val, channel_uuid) pssh_data, default_kid = get_pssh_from_mpd(dash_url) output = { "Cdn": [], "ManifestUrl": dash_url, "Headers": { "Manifest": {'User-Agent': user_agent, 'Referer': 'https://tvsmart.vectra.pl/'}, "Media": {'User-Agent': user_agent, 'Referer': 'https://tvsmart.vectra.pl/'} }, "Heartbeat": { "Url": '', "Params": '', "PeriodMs": 5*60*1000 }, "LicenseUrl": licence_url, "SessionId": session_id_val, "VideoId": video_id } if pssh_data: output['Pssh'] = pssh_data if default_kid: output['DefaultKid'] = default_kid print(json.dumps(output)) delete_session(token, session_id_val) except Exception as e: print(f"Error getting manifest: {e}", file=sys.stderr) return "error" elif action == "cdm" and cdm == "internal": licence_url = o11.parse_params(sys.argv, 'licenseUrl') session_id_val = o11.parse_params(sys.argv, 'sessionId') video_id = o11.parse_params(sys.argv, 'videoId') try: lic_headers = { 'origin': 'https://tvsmart.vectra.pl', 'referer': 'https://tvsmart.vectra.pl/', 'content-type': 'application/x-www-form-urlencoded', } params = {'videoAssetId': video_id, 'videoSessionId': session_id_val} response = req.post(licence_url, params=params, 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": channel_uuid = id try: session_id_val, video_id = get_video_session(token, channel_uuid) dash_url, licence_url = get_single(session_id_val, channel_uuid) pssh_data, default_kid = get_pssh_from_mpd(dash_url) if pssh_data: keys = do_cdm_external(pssh_data, licence_url, session_id_val, video_id) # If we have a default KID, use it instead if default_kid and keys: keys = [f"{default_kid}:{k.split(':')[1]}" for k in keys] for key in keys: print(key) else: print("No PSSH found in manifest", file=sys.stderr) return "error" delete_session(token, session_id_val) 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()