#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import base64 import uuid import random import hmac import hashlib from urllib.parse import unquote 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 = 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 = '/PolsatBoxGo_' + user + '.tokens' user_agent = 'pbg_mobile_android_native_cpplayer/20900' def generate_message_id(): random_uuid = str(uuid.uuid4()) uuid_prefix = random_uuid[:8] numeric_identifier = str(random.randint(1, 999)).zfill(3) return f"{uuid_prefix}-{numeric_identifier}" def get_current_time(): current_time_utc = datetime.datetime.now(datetime.timezone.utc) return current_time_utc.strftime("%Y-%m-%dT%H:%M:%SZ") def generate_session_token(method, method_namespace, session): session_id = session['id'] expiration_time = session['keyExpirationTime'] key = session['key'] str3 = f"{session_id}|{expiration_time}|{method}|{method_namespace}" key_bytes = base64.urlsafe_b64decode(key + '=' * (-len(key) % 4)) mac = hmac.new(key_bytes, msg=str3.encode('utf-8'), digestmod=hashlib.sha256) digest = mac.digest() encoded_digest = base64.urlsafe_b64encode(digest).decode('utf-8') return f"{str3}|{encoded_digest}" def do_cdm_external(pssh_data, media_id, key_id, source_id, session): 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 = {'accept': '*/*', 'content-type': 'text/plain;charset=UTF-8', 'origin': 'https://polsatboxgo.pl', 'referer': 'https://polsatboxgo.pl/', 'user-agent': user_agent} data = json.dumps({"id": 1, "jsonrpc": "2.0", "method": "getWidevineLicense", "params": {"cpid": 0, "keyId": key_id, "mediaId": media_id, "object": base64.b64encode(challenge_data).decode(), "offline": False, "sourceId": source_id, "authData": {"sessionToken": generate_session_token('drm', 'getWidevineLicense', session)}, "message": {"id": generate_message_id(), "timestamp": get_current_time()}, "userAgentData": {"application": "native", "build": 20900, "deviceType": "mobile", "os": "android", "player": "cpplayer", "portal": "pbg", "widevine": True}}}).replace(' ', '') licence = req.post('https://b2c-www.redefine.pl/rpc/drm/', headers=lic_headers, data=data) resp_data = licence.json() if 'error' in resp_data: return [] lic = resp_data['result']['object']['license'] cdm_obj.parse_license(session_id, base64.b64decode(lic)) 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): headers = {'Accept': '*/*', 'Origin': 'https://polsatboxgo.pl', 'Referer': 'https://polsatboxgo.pl/', 'User-Agent': user_agent} 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': pssh_tag = cp.find('cenc:pssh') if pssh_tag: return pssh_tag.text return None def get_single(media_id, session): headers = {'User-Agent': user_agent, 'Content-Type': 'application/json; charset=UTF-8'} data = json.dumps({"id": 1, "jsonrpc": "2.0", "method": "prePlayData", "params": {"cpid": 0, "mediaId": media_id, "authData": {"sessionToken": generate_session_token('navigation', 'prePlayData', session)}, "message": {"id": generate_message_id(), "timestamp": get_current_time()}, "userAgentData": {"application": "native", "build": 20900, "deviceType": "mobile", "os": "android", "player": "cpplayer", "portal": "pbg", "widevine": True}}}).replace(' ', '') response = req.post('http://b2c-mobile.redefine.pl/rpc/navigation/', headers=headers, data=data) resp_data = response.json() source = resp_data['result']['mediaItem']['playback']['mediaSources'][-1] return source['id'], source['keyId'], source['url'] def get_channels(session): headers = {'accept': 'application/json', 'content-type': 'application/json;charset=UTF-8', 'origin': 'https://polsatboxgo.pl', 'referer': 'https://polsatboxgo.pl/', 'user-agent': user_agent} data = json.dumps({"id": 1, "jsonrpc": "2.0", "method": "getTvChannels", "params": {"offset": 0, "filters": [], "userAgentData": {"application": "native", "build": 20900, "deviceType": "mobile", "os": "android", "player": "cpplayer", "portal": "pbg", "widevine": True}, "authData": {"sessionToken": generate_session_token('navigation', 'getTvChannels', session)}}}).replace(' ', '') response = req.post('https://b2c-www.redefine.pl/rpc/navigation/', headers=headers, data=data) return response.json()['result']['results'] def do_login_request(username, pwd): headers = {'User-Agent': user_agent, 'Content-Type': 'application/json; charset=UTF-8'} data = json.dumps({"id": 1, "jsonrpc": "2.0", "method": "login", "params": {"authData": {"authProvider": "native", "deviceId": {"type": "other", "value": "2d33b5e94a1bc758"}, "login": username, "password": pwd}, "message": {"id": generate_message_id(), "timestamp": get_current_time()}, "userAgentData": {"application": "native", "build": 20900, "deviceType": "mobile", "os": "android", "player": "cpplayer", "portal": "pbg", "widevine": True}}}).replace(' ', '') response = req.post('https://b2c-mobile.redefine.pl/rpc/auth/login/', headers=headers, data=data) return response.json()['result']['session'] def login(): print("logging in...", file=sys.stderr) session = do_login_request(user, password) auth_data = {'session': session} 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)) session = auth['session'] except: return "error" if action == "channels": output = {'Channels': []} for c in get_channels(session): output['Channels'].append({'Name': c['title'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + c['id'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + c['id'], 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in get_channels(session): output['Events'].append({'Name': c['title'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'id=' + c['id'], 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'id=' + c['id'], '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": media_id = id.replace('id=', '') if id.startswith('id=') else id source_id, key_id, url = get_single(media_id, session) pssh_data = get_pssh_from_mpd(url) output = {"Cdn": [], "ManifestUrl": url, "Headers": {"Manifest": {'User-Agent': user_agent}, "Media": {'User-Agent': user_agent}}, "MediaId": media_id, "SourceId": source_id, "KeyId": key_id} if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) elif action == "cdm" and cdm == "external": media_id = id.replace('id=', '') if id.startswith('id=') else id source_id, key_id, url = get_single(media_id, session) pssh_data = get_pssh_from_mpd(url) if pssh_data: for key in do_cdm_external(pssh_data, media_id, key_id, source_id, session): print(key) if do_action() == "error": login() do_action()