import sys, random import json import time import xml.etree.ElementTree as ET import cloudscraper import cfscrape from datetime import date import requests import base64 import binascii from urllib.parse import urlencode from urllib3.util import connection, parse_url from dns import message, rdatatype, resolver from requests.adapters import HTTPAdapter from requests.packages.urllib3.poolmanager import PoolManager from requests_toolbelt.adapters.socket_options import SocketOptionsAdapter from io import BytesIO from http.cookies import SimpleCookie import ipaddress import gzip import requests import urllib import http import ssl import socket import socks import base64 import os import cloudscraper from urllib3.util import connection, parse_url import ssl c = cfscrape.create_scraper() CIPHER_POOL = [ 'ECDHE-ECDSA-AES128-GCM-SHA256', 'ECDHE-RSA-AES128-GCM-SHA256', 'ECDHE-ECDSA-AES256-GCM-SHA384', 'ECDHE-RSA-AES256-GCM-SHA384', 'ECDHE-ECDSA-CHACHA20-POLY1305', 'ECDHE-RSA-CHACHA20-POLY1305', 'DHE-RSA-AES128-GCM-SHA256', 'DHE-RSA-AES256-GCM-SHA384' ] class Custom_Adapter(HTTPAdapter): def __init__(self, *args, **kwargs): self.worker = kwargs.pop('worker', None) super().__init__(*args, **kwargs) def init_poolmanager_original(self, connections, maxsize, block=False, ssl_version=None, source_address=(), socket_options=[]): if ssl_version: ssl_context = ssl.create_default_context() ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2 ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2 ssl_context.options = ssl.PROTOCOL_TLS & ssl.OP_NO_TLSv1_3 self.poolmanager = PoolManager(num_pools=connections, maxsize=maxsize, block=block, source_address=source_address, socket_options=socket_options, ssl_context=ssl_context) else: self.poolmanager = PoolManager(num_pools=connections, maxsize=maxsize, block=block, source_address=source_address, socket_options=socket_options) def init_poolmanager(self, connections, maxsize, block=False, ssl_version=None, source_address=(), socket_options=[]): # Randomize cipher selection random.shuffle(CIPHER_POOL) selected_ciphers = CIPHER_POOL[:random.randint(2, len(CIPHER_POOL))] cipher_string = ':'.join(selected_ciphers) ssl_context = ssl.create_default_context() ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2 ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2 try: ssl_context.set_ciphers(cipher_string) except ssl.SSLError: # Fallback to safe default if randomization fails ssl_context.set_ciphers('ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256') ssl_context.options |= ssl.OP_NO_COMPRESSION ssl_context.set_alpn_protocols(['h2', 'http/1.1']) self.poolmanager = PoolManager( num_pools=connections, maxsize=maxsize, block=block, source_address=source_address, socket_options=socket_options, ssl_context=ssl_context, ) def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None): parsed = parse_url(request.url) if self.worker and parsed.scheme in ('http', 'https'): request.headers['original-host'] = parsed.host if parsed.query: new_url = f"{parsed.scheme}://{self.worker}{parsed.path}?{parsed.query}" else: new_url = f"{parsed.scheme}://{self.worker}{parsed.path}" request.url = new_url return super().send(request, stream, timeout, verify, cert, proxies) class Session1: def __init__(self, bind="", proxy="", worker="", force_tls1_2=False, cloud=False): # Initialize instance variables self.session = None # Create session object if cloud: self.session = cloudscraper.create_scraper() else: self.session = requests.Session() # Configure proxies if proxy: self.session.proxies = {"http": proxy, "https": proxy} # Configure adapters only for non-cloud sessions if not cloud: self.session.mount('http://', Custom_Adapter(worker=worker)) self.session.mount('https://', Custom_Adapter(worker=worker)) # Configure network binding if bind: if "." not in bind: # Interface name binding socket_options = [(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, bind.encode())] source_address = ("", 0) else: # IP address binding socket_options = [] source_address = (bind, 0) else: source_address = ("", 0) socket_options = [] # Configure TLS version ssl_version = ssl.PROTOCOL_TLSv1_2 if force_tls1_2 else None # Configure pool managers http_adapter = self.session.get_adapter('http://') https_adapter = self.session.get_adapter('https://') http_adapter.init_poolmanager( connections=requests.adapters.DEFAULT_POOLSIZE, maxsize=requests.adapters.DEFAULT_POOLSIZE, source_address=source_address, socket_options=socket_options ) https_adapter.init_poolmanager( connections=requests.adapters.DEFAULT_POOLSIZE, maxsize=requests.adapters.DEFAULT_POOLSIZE, source_address=source_address, socket_options=socket_options, ssl_version=ssl_version ) def get_session(self): return self.session # Usage b = Session1(bind="", proxy="", worker="", force_tls1_2=True, cloud=False) DRM_SCHEME = "playready" # Change to "widevine" if needed req = b.get_session() def get_token(email, password, proxy_dict): url = "https://authentication-prod.ar.indazn.com/v5/SignIn" payload = { "Email": email, "Password": password, "Platform": "tv", "DeviceId": "00346aeb01" } headers = { "accept": "*/*", "accept-encoding": "gzip, deflate, br, zstd", "content-type": "application/json", "origin": "https://www.dazn.com", "User-Agent": USER_AGENT, "referer": "https://www.dazn.com/" } response = c.post(url, json=payload, headers=headers, proxies=proxy_dict) if response.status_code == 200: response_data = response.json() return response_data.get('AuthToken', {}).get('Token') else: print(f"Fehler beim Abrufen des Tokens: {response.status_code}") return None def get_pssh_from_mpd(mpd_url, proxy_dict): """Scarica la MPD e estrae il PSSH con il DNS corretto""" # MODIFICA IL DNS PRIMA DELLA RICHIESTA HTTP mpd_url = replace_links(mpd_url) headers = { "User-Agent": 'Mozilla/5.0 (Web0S; Linux/SmartTV) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.34 Safari/537.36 HbbTV/1.4.1 (+DRM; LGE; 65SM8200PLA; WEBOS4.5 05.30.10; W45_K5LP; DTV_W19P;) FVC/3.0 (LGE; WEBOS4.5 ;)' } response = c.get(mpd_url, proxies=proxy_dict, headers=headers) if response.status_code == 200: mpd_content = response.content root = ET.fromstring(mpd_content) namespaces = { 'mpd': 'urn:mpeg:dash:schema:mpd:2011', 'cenc': 'urn:mpeg:cenc:2013', 'mspr': 'urn:microsoft:playready' } # Cerca il PSSH for content_protection in root.findall('.//mpd:ContentProtection', namespaces): scheme_id_uri = content_protection.get('schemeIdUri') if scheme_id_uri and scheme_id_uri.lower() == "urn:uuid:9a04f079-9840-4286-ab92-e65be0885f95": pssh = content_protection.find('cenc:pssh', namespaces) if pssh is not None and pssh.text: return pssh.text print("PlayReady PSSH not found.") return None else: print(f"Failed to download MPD file: HTTP {response.status_code}") return None def request_decryption_keys(pssh, license_url, license_headers, token, proxy_dict): """Sendet den PSSH, die Lizenz-URL und den Token an die CDRM-API und gibt die Antwort zurück.""" host = "http://31.207.45.191:52317" key = '1e39b21d4f844ea793b922b9f82dd1bf' username = 'tutim' headers = {'x-api-key': key, 'x-api-username': username} json_data = {"pssh": pssh} res = c.post(url=f"{host}/cache", json=json_data, headers=headers, proxies=proxy_dict).json() if res.get("keys"): print("Got them from cache") Keys = [] for key_pair in res["keys"].split(" "): kid, key = key_pair.split(":") Keys.append(f"{kid}:{key}") Keys = " ".join(Keys) return Keys sys.stdout.write("") sys.stdout.flush() json_data = {"pssh": pssh, "device_name": "w4k"} res = c.post(url=f"{host}/challenge", json=json_data, headers=headers).json() challenge = res["challenge"] lic = req.post( url=license_url + f"&authToken={token}", headers={ "user-agent": USER_AGENT, "content-type": "text/xml; charset=utf-8", "accept": "*/*", "origin": "https://tv.dazn.com", "referer": "https://tv.dazn.com/", }, data=challenge, proxies=proxy_dict, ) print(lic.text) if lic.status_code == 200: license_ = lic.content else: print("Failed to get license") sys.exit(1) if isinstance(license_, bytes): license_ = base64.b64encode(license_).decode() license_ = base64.b64decode(license_).decode() json_data = {"pssh": pssh, "device_name": "w4k", "license": license_} res = c.post(url=f"{host}/keys", json=json_data, headers=headers, proxies=proxy_dict).json() Keys = [] for key_pair in res["keys"].split(" "): kid, key = key_pair.split(":") Keys.append(f"{kid}:{key}") Keys = " ".join(Keys) print(Keys) return Keys # Funzione per sostituire i link def replace_links(mpd_url): old_strings = [ "https://dca-fs-live-dazn-cdn.dazn.com", "https://dcblivedazn.akamaized.net", "https://dca-deit-livedazn.daznedge.net", "https://dcb-deit-livedazn.daznedge.net", "https://dcblivedazn.akamaized.net", "https://dca-ac-live.cdn.indazn.com", "https://dcb-gc-live.cdngc.dazn.com", "https://dcaos-de-livedazn.daznedge.net", "https://dcb-fs-live-dazn-cdn.dazn.com", "https://dcb-ak-livedazn.akamaized.net", "https://dca-ak-livedazn.akamaized.net", "https://dcb-ac-live.cdn.indazn.com" ] new_string = "https://dca-tm-livedazn.dazn.ticdn.it" for old_str in old_strings: mpd_url = mpd_url.replace(old_str, new_string) return mpd_url # Parte principale dello script if __name__ == "__main__": USER_AGENT = 'Mozilla/5.0 (Web0S; Linux/SmartTV) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.34 Safari/537.36 HbbTV/1.4.1 (+DRM; LGE; 65SM8200PLA; WEBOS4.5 05.30.10; W45_K5LP; DTV_W19P;) FVC/3.0 (LGE; WEBOS4.5 ;)' proxy_url = "" country_code = "it" language_code = "it" time_offset = 60 noLinearChannels = True pin = "0000" email = "servicecenterpay@gmail.com" password = "Lcdsrn869q" proxy_dict = {"http": proxy_url, "https": proxy_url} token = get_token(email, password, proxy_dict) print(token) try: try: with open("authorisationToken.txt") as ftoken: token = ftoken.read() except FileNotFoundError: if token: with open("authorisationToken.txt", "w") as f: f.write(token) else: print('LOGIN FAILED') exit() today = date.today() url = f"https://epg.discovery.indazn.com/eu/v2/Epg?date={today.year}-{today.month:02d}-{today.day:02d}&country={country_code}&languageCode={language_code}&openBrowse=false&timeZoneOffset={time_offset}" with requests.get(url) as responseData: response = responseData data = response.json() output = [] for tile in data.get('Tiles', []): VideoType = tile.get('VideoType') IsLinear = tile.get('IsLinear') competition = tile.get('Competition', [{}]).get("Title") LaUrl = data.get("PlaybackDetails", [{}])[0].get("LaUrl") if ((noLinearChannels and VideoType == "Live" and not IsLinear) or ( not noLinearChannels and VideoType == "Live")): canali = {} canali['asset_id'] = tile.get('AssetId') canali['start'] = tile.get('Start') canali['end'] = tile.get('End') canali['imageurl'] = f"https://image.discovery.indazn.com/jp/v3/jp/none/{tile['Image']['Id']}/fill/center/top/none/85/1920/1083/webp/image" output.append(canali) channels = [] count = 1 for channel in output: url = "https://api.playback.indazn.com/v5/Playback" querystring = { "AppVersion": "0.60.2", "DrmType": "PLAYREADY", "Format": "MPEG-DASH", "PlayerId": "@dazn/peng-html5-core/tv-next/tv", "Platform": "mob", "LanguageCode": "it", "Model": "unknown", "Secure": "true", "Manufacturer": "lg", "PlayReadyInitiator": "false", "Capabilities": "mta,hevc,hdr", "AssetId": '%s' % channel['asset_id'], "MtaLanguageCode": "" } headers = { "x-correlation-id": "92a88d22-111b-4a97-a9ff-d5718558c88b", "x-dazn-device": "523ee9a6-066a-4e7e-a3be-4d6d94696842", "content-type": "application/json", "Authorization": f"Bearer {token}", "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "User-Agent": USER_AGENT, 'x-age-verification-pin': pin, } with req.get(url, headers=headers, params=querystring, proxies=proxy_dict) as responseData: if 'Playback not allowed' in responseData.text: print("CFSCRAPE probably banned") sys.exit(1) response = responseData needed_data = {} if response.status_code == 401: headers_refresh = { 'User-Agent': USER_AGENT, 'Content-Type': 'application/json', 'Authorization': 'Bearer %s' % token, } json_data_refresh = { 'DeviceId': '0000000000' } with requests.post('https://ott-authz-bff-prod.ar.indazn.com/v5/RefreshAccessToken', headers=headers_refresh, json=json_data_refresh, proxies=proxy_dict) as responseData: response_refresh = responseData data_refresh = response_refresh.json() token = data_refresh.get("AuthToken", {}).get("Token") with open("authorisationToken.txt", "w") as f: f.write(token) headers['Authorization'] = f"Bearer {token}" with req.get(url, headers=headers, params=querystring, proxies=proxy_dict) as responseData: response = responseData if response.status_code == 200: data = response.json() eventTitle = data.get("Asset", {}).get("Title") manifest_url = None license_url = None for playback in data.get('PlaybackDetails', []): if 'live' in playback['ManifestUrl']: manifest_url = f"{playback['ManifestUrl']}" license_url = playback['LaUrl'] needed_data = { "Manifest URL": manifest_url, "License URL": license_url } break if not needed_data: sys.exit(1) manifest_url = replace_links(needed_data['Manifest URL']) print(f"Manifest URL modificato: {manifest_url}") pssh = get_pssh_from_mpd(manifest_url, proxy_dict) if "PSSH not found" not in pssh: print(f"Found PSSH: {pssh}") license_headers = { 'User-Agent': USER_AGENT, 'Accept': '*/*', 'Origin': 'https://www.dazn.com', 'Referer': 'https://www.dazn.com/', 'Accept-Language': 'it-IT,de;q=0.9,en-US;q=0.8,en;q=0.7', 'Content-Type': 'text/xml; charset=UTF-8', 'SOAPAction': 'http://schemas.microsoft.com/DRM/2007/03/protocols/AcquireLicense', 'x-correlation-id': 'e6a75e52-92e5-41f8-b805-849f214924ff', 'Authorization': f"Bearer {token}", 'priority': 'u=1, i', 'sec-ch-ua': '"Chromium";v="110", "Not;A=Brand";v="24", "Google Chrome";v="110"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'cross-site', } print("Going to sleep 35 seconds before other channel... have patience please") time.sleep(35) decryption_message = request_decryption_keys(pssh, needed_data['License URL'], license_headers, token, proxy_dict) canale = { "Id": str(count), "Name": eventTitle, "mpdUrl": manifest_url, "startDate": channel['start'], "endDate": channel['end'], "imageurl": channel['imageurl'], "kid": decryption_message.split(":")[0], "key": decryption_message.split(":")[1], } channels.append(canale) count += 1 # Sostituisci i link in 'mpdUrl' nel JSON finale for channel in channels: channel['mpdUrl'] = replace_links(channel['mpdUrl']) with open("output.json", 'w+') as f: json.dump(channels, f, indent=2) channels_with_additional_info = [] for channel in channels: channel_with_info = { "Id": f"Dazn_{channels.index(channel) + 1}", "OriginalName": channel["Name"], "Name": channel["Name"], "Competizione": "Serie A Enilive", "Sport": "SportName", "mpdUrl": channel["mpdUrl"], "startDate": channel["startDate"], "endDate": channel["endDate"], "imageurl": channel["imageurl"], "kid": channel["kid"], "key": channel["key"], "XacellURL": replace_links(channel["mpdUrl"]) + "&decryption_key=" + channel["kid"] + ":" + channel["key"], "XacellURL DNS aggiornato": replace_links(channel["mpdUrl"]) + "&decryption_key=" + channel["kid"] + ":" + channel["key"], } channels_with_additional_info.append(channel_with_info) with open("daznevent.json", 'w+') as f: json.dump(channels_with_additional_info, f, indent=2) except Exception as e: print("An error occurred:", str(e))