#!/usr/bin/env python3 """ ==================================================================== DAZN - Local HTTPS Proxy (aiohttp) Proxy locale che inietta automaticamente header e token DAZN. Permette di usare player standard (VLC, mpv) fornendo solo l'URL locale e la chiave ClearKey. Modificato per HTTPS e aiohttp. ==================================================================== """ import os import sys import json import time import urllib.parse import socket import ssl import subprocess import asyncio import re try: from aiohttp import web except ImportError: print("ERRORE: aiohttp non รจ installato. Esegui: pip install aiohttp", flush=True) sys.exit(1) try: from curl_cffi import requests GLOBAL_SESSION = requests.Session(impersonate="chrome120") except ImportError: import requests GLOBAL_SESSION = requests.Session() # Colori per il log GREEN = '\033[32m' YELLOW = '\033[33m' RED = '\033[31m' CYAN = '\033[36m' RESET = '\033[0m' def log(msg: str, color: str = RESET) -> None: print(f"{color}[{time.strftime('%H:%M:%S')}] [PROXY] {msg}{RESET}", flush=True) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) CHANNELS_FILE = os.path.join(SCRIPT_DIR, "dazn_channels.json") CONFIG_FILE = os.path.join(SCRIPT_DIR, "dazn_config.json") _channels_cache = [] _channels_mtime = 0 def load_channels(): global _channels_cache, _channels_mtime if os.path.exists(CHANNELS_FILE): try: mtime = os.path.getmtime(CHANNELS_FILE) if mtime > _channels_mtime: with open(CHANNELS_FILE, 'r', encoding='utf-8') as f: _channels_cache = json.load(f).get("channels", []) _channels_mtime = mtime except Exception as e: log(f"Errore lettura dazn_channels.json: {e}", RED) return _channels_cache _headers_cache = {} _headers_mtime = 0 def load_headers(): global _headers_cache, _headers_mtime default_headers = { "Origin": "https://www.dazn.com", "Referer": "https://www.dazn.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36", "sec-ch-ua": '"Chromium";v="152", "Not?A_Brand";v="24", "Google Chrome";v="152"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"' } if os.path.exists(CONFIG_FILE): try: mtime = os.path.getmtime(CONFIG_FILE) if mtime > _headers_mtime: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: cfg = json.load(f) h = cfg.get("dazn", {}).get("headers", {}) for k, v in h.items(): default_headers[k] = v _headers_cache = default_headers _headers_mtime = mtime except Exception: pass return _headers_cache if _headers_cache else default_headers def fetch_sync(url, headers, is_mpd): kwargs = {'headers': headers, 'timeout': 10} if not is_mpd: kwargs['stream'] = True return GLOBAL_SESSION.get(url, **kwargs) async def handle_playlist(request): channels = load_channels() lines = ["#EXTM3U x-tvg-url=\"\""] host = request.host if not host: host = "addictedbytheproject.nl:10443" for ch in channels: if ch.get("status") == "OK": lines.append(f'#EXTINF:-1 tvg-id="{ch.get("id")}" tvg-name="{ch.get("name")}" group-title="DAZN",{ch.get("name")}') lines.append('#KODIPROP:inputstream=inputstream.adaptive') lines.append('#KODIPROP:inputstream.adaptive.manifest_type=mpd') lines.append('#KODIPROP:inputstream.adaptive.license_type=clearkey') lines.append(f'#KODIPROP:inputstream.adaptive.license_key={ch.get("key_string")}') # Usa https:// per il dominio lines.append(f'https://{host}/play/{ch.get("id")}/stream.mpd') body = ("\n".join(lines) + "\n") log("Generata playlist locale m3u", CYAN) return web.Response(text=body, content_type='audio/mpegurl', headers={'Access-Control-Allow-Origin': '*'}) async def handle_play(request): channel_id = request.match_info.get('id') filename = request.match_info.get('filename') if not channel_id or not filename: raise web.HTTPBadRequest(text="Bad Request") channels = load_channels() channel = next((c for c in channels if c.get("id") == channel_id), None) if not channel: raise web.HTTPNotFound(text="Channel Not Found") base_manifest_url = channel.get("base_manifest_url") cdn_token = channel.get("cdn_token", {}) token_name = cdn_token.get("Name", "dazn-token") token_val = cdn_token.get("Value", "") if not base_manifest_url or not token_val: raise web.HTTPInternalServerError(text="Channel data incomplete") parsed_base = urllib.parse.urlparse(base_manifest_url) base_path = parsed_base.path.rsplit('/', 1)[0] base_cdn_url = f"{parsed_base.scheme}://{parsed_base.netloc}{base_path}/" dazn_headers = load_headers() dazn_headers["Cookie"] = f"{token_name}={token_val}" is_mpd = filename.startswith("stream.mpd") if is_mpd: target_url = base_manifest_url if "?" in target_url: target_url += f"&{token_name}={token_val}" else: target_url += f"?{token_name}={token_val}" else: clean_filename = filename.split('?')[0] target_url = f"{base_cdn_url}{clean_filename}?{token_name}={token_val}" req_headers = dazn_headers.copy() if 'Host' in req_headers: del req_headers['Host'] try: r = await asyncio.to_thread(fetch_sync, target_url, req_headers, is_mpd) if not r.ok: if is_mpd: log(f"Errore CDN ({r.status_code}) per {filename}", RED) raise web.HTTPBadGateway(text=f"CDN Error: {r.status_code}") if is_mpd: mpd_text = r.content.decode('utf-8', errors='ignore') mpd_text = re.sub(r']*schemeIdUri="urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed"[^>]*>.*?', '', mpd_text, flags=re.DOTALL) mpd_text = re.sub(r']*schemeIdUri="urn:microsoft:playready"[^>]*>.*?', '', mpd_text, flags=re.DOTALL) mpd_text = re.sub(r']*schemeIdUri="urn:uuid:9a04f079-9840-4286-ab92-e65be0885f95"[^>]*>.*?', '', mpd_text, flags=re.DOTALL) clearkey_cp = '' mpd_text = re.sub(r'(]*value="cenc"[^>]*/>)', r'\1' + clearkey_cp, mpd_text) body = mpd_text.encode('utf-8') resp = web.Response(body=body, status=r.status_code, headers={'Access-Control-Allow-Origin': '*'}) if 'Content-Type' in r.headers: resp.headers['Content-Type'] = r.headers['Content-Type'] log(f"MPD scaricato, patchato (ClearKey) e proxiato per {channel.get('name')}", GREEN) return resp else: resp = web.StreamResponse(status=r.status_code, headers={'Access-Control-Allow-Origin': '*'}) if 'Content-Type' in r.headers: resp.headers['Content-Type'] = r.headers['Content-Type'] if 'Content-Length' in r.headers: resp.headers['Content-Length'] = r.headers['Content-Length'] await resp.prepare(request) # Sincrono over aiohttp for chunk in r.iter_content(chunk_size=65536): if chunk: await resp.write(chunk) await resp.write_eof() return resp except web.HTTPException: raise except Exception as e: log(f"Proxy Error: {e}", RED) raise web.HTTPInternalServerError(text="Internal Proxy Error") finally: if 'r' in locals() and hasattr(r, 'close'): r.close() def generate_cert(): cert_path = os.path.join(SCRIPT_DIR, "cert.pem") key_path = os.path.join(SCRIPT_DIR, "key.pem") if not os.path.exists(cert_path) or not os.path.exists(key_path): log("Generazione di un nuovo certificato SSL self-signed via OpenSSL...", YELLOW) try: subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key_path, "-out", cert_path, "-days", "365", "-subj", "/CN=localhost", "-addext", "subjectAltName=DNS:addictedbytheproject.nl,DNS:localhost"], check=True) log("Certificato SSL generato con successo.", GREEN) except Exception as e: log(f"Errore durante la generazione del certificato SSL: {e}", RED) return cert_path, key_path def init_app(): app = web.Application() app.router.add_get('/', handle_playlist) app.router.add_get('/playlist.m3u', handle_playlist) app.router.add_get('/play/{id}/{filename:.*}', handle_play) return app if __name__ == '__main__': port = 10443 log(f"Avvio DAZN HTTPS Proxy Server sulla porta {port}...", CYAN) cert_path, key_path = generate_cert() ssl_context = None if os.path.exists(cert_path) and os.path.exists(key_path): try: ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_context.load_cert_chain(cert_path, key_path) log("Contesto SSL inizializzato correttamente.", GREEN) except Exception as e: log(f"Errore inizializzazione SSL: {e}", RED) ssl_context = None try: # bind a 0.0.0.0 web.run_app(init_app(), host='0.0.0.0', port=port, ssl_context=ssl_context, print=None) except KeyboardInterrupt: log("\nArresto Proxy Server...", YELLOW) sys.exit(0)