#!/usr/bin/python3 import sys import os import o11 import json import datetime import pytz import base64 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') 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_param = 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 = '/CanalPlusPL_' + user + '.tokens' 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' user_key_id = '_acbpff1ya' device_key_id = '1629342960819-9acbegg0d451' def do_cdm_external(pssh_data, epg_id, live_token, loc): 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) payload = base64.b64encode(challenge_data).decode() lic_headers = {'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': 'https://www.canalplus.com', 'referer': 'https://www.canalplus.com/', 'user-agent': user_agent} data = {'ServiceRequest': {'InData': {'EpgId': epg_id, 'LiveToken': live_token, 'UserKeyId': user_key_id, 'DeviceKeyId': device_key_id, 'ChallengeInfo': payload, 'Mode': 'MKPL'}}} licence = req.post(f'https://ltv.slc-app-aka.prod.bo.canal.canalplustech.pro/api/V4/zones/cp{loc}/devices/31/apps/1/jobs/GetLicence', headers=lic_headers, json=data) resp = licence.json() cdm_obj.parse_license(session_id, base64.b64decode(resp['ServiceResponse']['OutData']['LicenseInfo'])) 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 = {'origin': 'https://www.canalplus.com', 'referer': 'https://www.canalplus.com/', 'user-agent': user_agent} response = req.get(url, headers=headers) psshs = BeautifulSoup(response.content, features="xml").findAll('cenc:pssh') return psshs[1].text if len(psshs) > 1 else (psshs[0].text if psshs else None) def get_channels(pass_token, loc): headers = {'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': 'https://www.canalplus.com', 'referer': 'https://www.canalplus.com/', 'user-agent': user_agent} data = {'ServiceRequest': {'InData': {'UseRmuTokenPlaceHolder': True, 'PassData': {'Id': 0, 'Token': pass_token}, 'UserKeyId': user_key_id, 'DeviceKeyId': device_key_id, 'PDSData': {'GroupTypes': '1;2;4'}}}} response = req.post(f'https://ltv.slc-app-aka.prod.bo.canal.canalplustech.pro/api/V4/zones/cp{loc}/devices/3/apps/1/jobs/InitLiveTV', headers=headers, json=data) resp_data = response.json() out_data = resp_data['ServiceResponse']['OutData'] live_token = out_data['LiveToken'] rmu_token = out_data['RMUToken'] channels_group = out_data['PDS']['ChannelsGroups']['ChannelsGroup'] channels = [] for cg in channels_group: channels += cg['Channels'] return live_token, channels, rmu_token def get_channel_link(url, rmu_token): headers = {'accept': 'application/json, text/plain, */*', 'origin': 'https://www.canalplus.com', 'referer': 'https://www.canalplus.com/', 'user-agent': user_agent} params = {'token': rmu_token} response = req.get(url.split('?')[0], params=params, headers=headers) return response.json().get('primary', {}).get('src', '') def pass_token_request(pass_id, region): headers = {'content-type': 'application/x-www-form-urlencoded', 'origin': 'https://www.canalplus.com', 'referer': 'https://www.canalplus.com/', 'user-agent': user_agent} data = 'media=web&noCache=false&portailId=vbdTj7eb6aM.&passIdType=pass&vect=INTERNET&zone=cpfra&passId=' + urllib.parse.quote_plus(pass_id) response = req.post(f'https://pass-api-v2.canal-plus.com/provider/services/{region.lower()}/public/createToken', headers=headers, data=data) return response.json().get('response', {}).get('passToken', '') def get_loc(region): locs = {'FR': 'fra', 'PL': 'pol', 'GP': 'ant'} return locs.get(region, 'pol') def login(): print("Please set passId in token file manually", file=sys.stderr) auth_data = {'pass_id': ''} json.dump(auth_data, open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w')) print("Token file created - add your passId", file=sys.stderr) def get_auth(): auth = json.load(open(os.path.abspath(os.path.dirname(__file__)) + authFile)) pass_id = auth['pass_id'] region = pass_id[:2] pass_token = pass_token_request(pass_id, region) loc = get_loc(region) return pass_token, loc def do_action(): if action == "login": login() sys.exit() try: pass_token, loc = get_auth() except: return "error" live_token, channels, rmu_token = get_channels(pass_token, loc) channels = [c for c in channels if c.get('AccessRights', {}).get('NoAccess') == 'false'] if action == "channels": output = {'Channels': []} for c in channels: output['Channels'].append({'Name': c['Name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'epg=' + str(c['EpgId']) + '&wsx=' + urllib.parse.quote_plus(c['WSXUrl']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'epg=' + str(c['EpgId']) + '&wsx=' + urllib.parse.quote_plus(c['WSXUrl']), 'Video': 'best'}) print(json.dumps(output, indent=2)) elif action == "events": output = {'Events': []} for c in channels: output['Events'].append({'Name': c['Name'], 'Mode': "live", 'SessionManifest': True, 'ManifestScript': 'epg=' + str(c['EpgId']) + '&wsx=' + urllib.parse.quote_plus(c['WSXUrl']), 'CdmType': "widevine", 'UseCdm': True, 'Cdm': 'epg=' + str(c['EpgId']) + '&wsx=' + urllib.parse.quote_plus(c['WSXUrl']), '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": params = dict(p.split('=', 1) for p in id.split('&') if '=' in p) epg_id = params.get('epg', '') wsx_url = urllib.parse.unquote_plus(params.get('wsx', '')) url = get_channel_link(wsx_url, rmu_token) pssh_data = get_pssh_from_mpd(url) output = {"Cdn": [], "ManifestUrl": url, "Headers": {"Manifest": {'User-Agent': user_agent, 'Origin': 'https://www.canalplus.com', 'Referer': 'https://www.canalplus.com/'}, "Media": {'User-Agent': user_agent, 'Origin': 'https://www.canalplus.com', 'Referer': 'https://www.canalplus.com/'}}, "EpgId": epg_id, "LiveToken": live_token, "Loc": loc} if pssh_data: output['Pssh'] = pssh_data print(json.dumps(output)) elif action == "cdm" and cdm_param == "external": params = dict(p.split('=', 1) for p in id.split('&') if '=' in p) epg_id = params.get('epg', '') wsx_url = urllib.parse.unquote_plus(params.get('wsx', '')) url = get_channel_link(wsx_url, rmu_token) pssh_data = get_pssh_from_mpd(url) if pssh_data: for key in do_cdm_external(pssh_data, epg_id, live_token, loc): print(key) if do_action() == "error": login() do_action()