#!/usr/bin/env python # -*- coding: utf-8 -*- from __future__ import absolute_import, print_function import base64 import gzip try: import requests except ImportError: # requests is a hard runtime dependency (python3-requests, declared # in CONTROL/control) - VavooProxy is built directly on # requests.Session() and cannot function without it, but a clear # message here beats a bare traceback on images missing the package. print("[PROXY] FATAL: 'requests' package not found - install python3-requests") raise # import uuid import time import threading import socket import select try: import urllib3 except ImportError: urllib3 = None import atexit import os from collections import OrderedDict from json import loads, load, dumps from . import ( PORT, PROXY_HOST, PROXY_STATUS_URL, PROXY_SHUTDOWN_URL, BASE_SITES, HOST_GIT, SREF_MAP_FILE, __version__ ) from .vUtils import ( _starting_lock, debug, get_external_ip, is_proxy_running, log_exception, make_print, trace_error, # RequestAgent ) _starting = False try: unicode except NameError: unicode = str print = make_print("PROXY") try: from urllib.parse import unquote, urlparse, parse_qs except ImportError: from urllib import unquote from urlparse import urlparse, parse_qs # Python 2/3 compatibility for exception names used in handlers try: BrokenPipeError except NameError: # Python 2 BrokenPipeError = IOError try: ConnectionResetError except NameError: # Python 2 ConnectionResetError = IOError try: ConnectionError except NameError: # Python 2 ConnectionError = IOError # Global stop flag used for clean shutdown (prevents restart loop) STOP_EVENT = threading.Event() # NOTE: do NOT call socket.setdefaulttimeout() globally – it poisons # streaming sockets and all other network ops in the same process. try: from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer print(" Python 2 detected") except ImportError: from http.server import BaseHTTPRequestHandler, HTTPServer print(" Python 3 detected") # Threaded HTTP server (prevents one streaming client from blocking others) try: # Py3 from socketserver import ThreadingMixIn except ImportError: # Py2 from SocketServer import ThreadingMixIn # Disable SSL warnings if urllib3 is not None: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True allow_reuse_address = True """ ######################################################### # # # Vavoo Stream Live Plugin # # Created by Lululla (https://github.com/Belfagor2005) # # License: CC BY-NC-SA 4.0 # # https://creativecommons.org/licenses/by-nc-sa/4.0 # # Last Modified: 202600503 # # # # Credits: # # - Original concept by Lululla # # - Background images by @oktus # # - Additional contributions by Qu4k3 # # - Linuxsat-support.com & Corvoboys communities # # # # Usage of this code without proper attribution # # is strictly prohibited. # # For modifications and redistribution, # # please maintain this credit header. # ######################################################### """ # ========== CONFIGURATIONS ========== """ ==================================================================== VAVOO PROXY API HELP ==================================================================== Proxy Base URL: http://127.0.0.1:4323 All endpoints are accessible locally from: - Enigma2 - wget - curl - VLC - ffplay - IPTV players ==================================================================== 1. /status - Proxy Status ==================================================================== URL: http://127.0.0.1:4323/status Description: Returns current proxy runtime status. Includes: - proxy initialization state - loaded channels count - addonSig validity - addonSig age - local IP address - listening port Example: wget -qO- http://127.0.0.1:4323/status Example Response: { "initialized": true, "channels_count": 1250, "addon_sig_valid": true, "addon_sig_age": 42, "local_ip": "192.168.1.10", "port": 4323 } ==================================================================== 2. /health - Health Check ==================================================================== URL: http://127.0.0.1:4323/health Description: Returns detailed health and monitoring information. Includes: - overall health status - token validity - token TTL - uptime - heartbeat age - local IP - listening port Important: This endpoint is READ-ONLY. It does NOT restart or refresh the proxy. Example: wget -qO- http://127.0.0.1:4323/health ==================================================================== 3. /countries - Available Countries ==================================================================== URL: http://127.0.0.1:4323/countries Description: Returns all unique countries available in the catalog. Excluded: - empty country names - "default" Example: wget -qO- http://127.0.0.1:4323/countries Example Response: [ "France", "Germany", "Italy", "Spain" ] ==================================================================== 4. /channels?country=CountryName - Channels By Country ==================================================================== URL: http://127.0.0.1:4323/channels?country=Italy Description: Returns all channels matching the specified country. Each channel contains: - id - name - logo - country - local playback URL Notes: - country matching is case-insensitive - country names should be URL encoded Example: wget -qO- "http://127.0.0.1:4323/channels?country=Italy" Example Response: [ { "id": "abc123", "name": "RAI 1", "url": "http://192.168.1.10:4323/vavoo?channel=abc123", "logo": "https://logo.png", "country": "Italy" } ] ==================================================================== 5. /catalog - Full Catalog ==================================================================== URL: http://127.0.0.1:4323/catalog Description: Returns the complete filtered catalog currently loaded in memory. Contains: - all channels - metadata - original stream information Example: wget -qO- http://127.0.0.1:4323/catalog ==================================================================== 6. /vavoo?channel=ChannelID - Stream Redirect ==================================================================== URL: http://127.0.0.1:4323/vavoo?channel=abc123 Description: Resolves the upstream stream URL and immediately returns a HTTP 302 redirect. Optimized for: - Enigma2 IPTV bouquets - VLC - ffplay - gstplayer - exteplayer3 - serviceapp Behavior: Client receives direct upstream stream URL. Example: wget -S -O /dev/null "http://127.0.0.1:4323/vavoo?channel=abc123" Expected Response: HTTP/1.1 302 Found ==================================================================== 7. /stream?ref=ServiceReference - Direct Stream Proxy ==================================================================== URL: http://127.0.0.1:4323/stream?ref=1%3A0%3A1%3A... Description: Streams MPEG-TS content directly through the proxy instead of redirecting the client. Features: - upstream buffering - keep-alive - timeout monitoring - chunked streaming - disabled caching Workflow: 1. Decode service reference 2. Map reference to channel ID 3. Resolve upstream stream 4. Proxy TS stream to client Recommended for: - Enigma2 service references - local TS proxying - IPTV middleware - transcoding pipelines Example: ffplay "http://127.0.0.1:4323/stream?ref=1%3A0%3A1%3A..." ==================================================================== 8. /refresh_token - Refresh addonSig ==================================================================== URL: http://127.0.0.1:4323/refresh_token Description: Forces addonSig token refresh. Returns: - success - error Example: wget -qO- http://127.0.0.1:4323/refresh_token Example Response: { "status": "success", "message": "Token refreshed" } ==================================================================== 9. /epg/.xml - EPG Redirect ==================================================================== URL: http://127.0.0.1:4323/epg/it.xml Description: Redirects to XMLTV EPG file hosted on GitHub. Examples: /epg/it.xml /epg/fr.xml /epg/de.xml Response: HTTP 302 Redirect Example: wget -O epg.xml "http://127.0.0.1:4323/epg/it.xml" ==================================================================== 10. /shutdown - Graceful Shutdown ==================================================================== URL: http://127.0.0.1:4323/shutdown Description: Gracefully shuts down the proxy server. Behavior: - stops HTTP server - closes sockets - stops proxy threads - sets global stop event Example: wget -qO- http://127.0.0.1:4323/shutdown ==================================================================== USEFUL ENIGMA2 TEST COMMANDS ==================================================================== Check proxy status: wget -qO- http://127.0.0.1:4323/status Check health: wget -qO- http://127.0.0.1:4323/health List countries: wget -qO- http://127.0.0.1:4323/countries Get Italy channels: wget -qO- "http://127.0.0.1:4323/channels?country=Italy" Check redirect: wget -S -O /dev/null "http://127.0.0.1:4323/vavoo?channel=abc123" Check listening port: netstat -lntp | grep 4323 or: ss -lntp | grep 4323 ==================================================================== """ # API Endpoints TOKEN_ADDON_SIG = 600 # 10 minutes - TOKEN EXPIRES EVERY 10 MINUTES! TOKEN_REFRESH_AGE = 480 GEOIP_URL = "https://www.vavoo.tv/geoip" PING_URL = "https://www.vavoo.tv/api/app/ping" PING_URL2 = "https://www.vypn.net/api/app/ping" # Real VYPN app identity (package/version from its APK metadata), # matching a confirmed-working ping payload for this endpoint. VYPN_PACKAGE = "net.vypn.app" VYPN_VERSION = "1.4.1" PID_FILE = "/tmp/vavoo_proxy.pid" BOOTING_FILE = "/tmp/vavoo_proxy_booting" # Primary + mirror. Some regions get HTTP 451 from the primary. # HEADERS = { # "accept": "*/*", # # "user-agent": RequestAgent(), # "user-agent": "Mozilla/5.0 (X11; Linux armv7l) AppleWebKit/537.36", # "Accept-Encoding": "gzip, deflate", # } # NOTE: do NOT set "Connection": "close" here - self.session.headers is # applied to every request through this session (catalog pages, resolve # calls, and the /stream handler's upstream GET), so it would force a # fresh TCP/TLS handshake on all of them instead of reusing a pooled # connection from the HTTPAdapter configured below. HEADERS = { "accept": "*/*", "user-agent": "MediaHubMX/2", "Accept-Language": "de", "Accept-Encoding": "gzip, deflate", } # Ceiling on how long a "booting" marker is trusted. A crash, power loss, # or plugin/box restart during boot can leave this file behind forever; # without a staleness check every later start would think a boot is # permanently in progress. Kept well above a normal catalog load (usually # well under a minute) so it never interferes with a real, slow boot. MAX_BOOT_AGE = 180 def is_proxy_booting(): """Check if another proxy instance is currently starting up. Treats the marker as stale (and removes it) if it is older than MAX_BOOT_AGE seconds, so a leftover file from a crashed process can't make every future start think a boot is still running. """ if not os.path.exists(BOOTING_FILE): return False try: age = time.time() - os.path.getmtime(BOOTING_FILE) except OSError: return False if age > MAX_BOOT_AGE: print("[PROXY] Stale booting marker ({}s old), removing".format(int(age))) remove_booting_file() return False return True def write_booting_file(): with open(BOOTING_FILE, 'w') as f: f.write(str(os.getpid())) def remove_booting_file(): if os.path.exists(BOOTING_FILE): os.unlink(BOOTING_FILE) def decode_response(resp): """Decode gzip response if needed (Py2/3 compatible).""" if resp.content[:2] == b'\x1f\x8b': # gzip.decompress is Py3.2+ only; use GzipFile for Py2 compat try: raw = gzip.decompress(resp.content) except AttributeError: import io as _io with gzip.GzipFile(fileobj=_io.BytesIO(resp.content)) as gz: raw = gz.read() return loads(raw.decode('utf-8', 'ignore')) try: return resp.json() except ValueError: return loads(resp.content.decode('utf-8', 'ignore')) def _rewrite_addon_sig_ip(sig, client_ip): """Rewrite the client IP embedded inside a base64-encoded addonSig. The token is base64(JSON) where the JSON has a "data" field that is itself a JSON string carrying an "ips"/"ip" pair. Point those at our own public IP so the resolved stream matches where our requests actually come from, matching a confirmed-working reference implementation. Falls back to the original, unmodified sig if the token doesn't have the expected shape. """ try: padded = sig + '=' * (-len(sig) % 4) decoded = base64.b64decode(padded) if isinstance(decoded, bytes): decoded = decoded.decode('utf-8') sig_obj = loads(decoded) if not isinstance(sig_obj, dict) or "data" not in sig_obj: return sig data_obj = loads(sig_obj["data"]) current_ips = data_obj.get("ips") if not isinstance(current_ips, list): current_ips = [] data_obj["ips"] = [client_ip] + \ [ip for ip in current_ips if ip and ip != client_ip] if isinstance(data_obj.get("ip"), (str, unicode)): data_obj["ip"] = client_ip sig_obj["data"] = dumps(data_obj) new_sig = base64.b64encode(dumps(sig_obj).encode('utf-8')) if isinstance(new_sig, bytes): new_sig = new_sig.decode('ascii') return new_sig except Exception as e: print("[AddonSig] IP rewrite failed, keeping original sig: {}".format(e)) return sig def is_proxy_already_running(): """Check if another proxy instance is running via PID file""" try: with open(PID_FILE, 'r') as f: pid = int(f.read().strip()) # Check if process exists os.kill(pid, 0) return True except (IOError, OSError, ValueError): return False except Exception as e: print(str(e)) log_exception("is_proxy_already_running error", area="PROXY") return False def write_pid_file(): """Write current PID to file""" try: with open(PID_FILE, 'w') as f: f.write(str(os.getpid())) return True except Exception: return False def remove_pid_file(): """Remove PID file on exit""" try: if os.path.exists(PID_FILE): os.unlink(PID_FILE) except Exception: pass atexit.register(remove_pid_file) class VavooProxy(object): def __init__(self): self.session = requests.Session() self.session.headers.update(HEADERS) # 1. ADAPTER IMPROVED: more intelligent retries adapter = requests.adapters.HTTPAdapter( pool_connections=20, pool_maxsize=20, max_retries=2, pool_block=False ) self.session.mount('http://', adapter) self.session.mount('https://', adapter) self.session.verify = False self.session.timeout = 60 self.session.stream = True # 2. REPLACE request wrapper with safer version self.session.request = self._robust_request self.active_streams = 0 self._stream_lock = threading.Lock() # guards active_streams counter self.addon_sig_data = {"sig": None, "ts": 0} self.addon_sig_lock = threading.Lock() self.all_filtered_items = [] self.channels_by_country = {} self.channels_by_id = {} self.countries_list = [] self.current_language = "en" self.current_region = "US" self.external_ip = None self.initialized = False self.last_heartbeat = time.time() self.local_ip = None self.refresh_timer = None # ordered for deterministic LRU eviction (Py2+3) self.resolve_cache = OrderedDict() self.resolve_cache_lock = threading.Lock() # stream URLs valid for ~5min; was 30s (too aggressive) self.resolve_cache_ttl = 300 self.server = None self.start_time = time.time() # Stop flag for background workers self._stop_event = threading.Event() self._token_monitor_thread = None # Mirror-aware endpoints self.base_sites = list(BASE_SITES) self.base_site_index = 0 self._update_endpoints() # Start lightweight token monitor only self.start_token_monitor() print(" Initialized at " + time.ctime()) def stream_started(self): with self._stream_lock: self.active_streams += 1 count = self.active_streams print("[Proxy] Stream started. Active streams: {}".format(count)) def stream_ended(self): with self._stream_lock: if self.active_streams > 0: self.active_streams -= 1 count = self.active_streams print("[Proxy] Stream ended. Active streams: {}".format(count)) def _update_endpoints(self): """Update API endpoints from the current base site.""" base = self.base_sites[self.base_site_index].rstrip('/') self.catalog_url = base + "/mediahubmx-catalog.json" self.resolve_url = base + "/mediahubmx-resolve.json" def _switch_to_next_base(self, reason=""): """Switch to next mirror base site.""" if not self.base_sites: return old = self.base_sites[self.base_site_index] self.base_site_index = (self.base_site_index + 1) % len(self.base_sites) self._update_endpoints() new = self.base_sites[self.base_site_index] print( " Switching base site: {0} -> {1} {2}".format(old, new, reason)) def _robust_request(self, method, url, **kwargs): """Simplified and safer version""" # Set reasonable timeouts if 'timeout' not in kwargs: kwargs['timeout'] = (5, 15) # 5s connect, 15s read try: # SINGLE REQUEST, no infinite retries # Call the real Session.request, bypassing our override to avoid # recursion response = requests.Session.request( self.session, method, url, **kwargs) return response except (requests.exceptions.Timeout, socket.timeout) as e: print(" Timeout on " + str(url) + ": " + str(e)) raise except requests.exceptions.ConnectionError as e: print(" Connection error on " + str(url) + ": " + str(e)) raise except Exception as e: print(" Error on " + str(url) + ": " + str(e)) raise def start_token_monitor(self): """Monitor token age with minimal background traffic""" def token_monitor_loop(): while not self._stop_event.is_set(): # Check FIRST, then sleep (first check is immediate at startup) try: now = time.time() token_age = (now - self.addon_sig_data["ts"] if self.addon_sig_data["sig"] else 0) if self.addon_sig_data["sig"] and token_age > TOKEN_REFRESH_AGE: print( "[Token Monitor] Token old ({}s), refreshing...".format( int(token_age))) self.refresh_addon_sig_if_needed(force=True) self.last_heartbeat = now except Exception as e: print("[Token Monitor] Error: " + str(e)) # Sleep interval: # - Streaming active → check every 30s (token must not expire mid-stream) # - Idle → check every 60s with self._stream_lock: streaming = self.active_streams > 0 if streaming: self._stop_event.wait(30) else: self._stop_event.wait(60) self._token_monitor_thread = threading.Thread( target=token_monitor_loop) self._token_monitor_thread.setDaemon(True) self._token_monitor_thread.start() print(" Token monitor started") def _get_cached_external_ip(self): """Public IP of this box, cached for the process lifetime. Used to rewrite the IP embedded in the addonSig token so it matches where our actual resolve/stream requests come from, instead of whatever the ping endpoint saw or guessed. """ if not self.external_ip: try: self.external_ip = get_external_ip() except Exception: self.external_ip = None return self.external_ip def refresh_addon_sig_if_needed(self, force=False): with self.addon_sig_lock: now = time.time() if not force and self.addon_sig_data["sig"] and ( now - self.addon_sig_data["ts"] < 300): return self.addon_sig_data["sig"] try: import uuid unique_id = str(uuid.uuid4()) current_timestamp = int(time.time() * 1000) # Matches a confirmed-working VYPN ping payload exactly - # the previous version carried extra fields (buildId, # engine, signatures, installer, devMode=True, ...) copied # from an older reverse-engineered client that this # endpoint no longer accepts as a real free-tier client. payload = { "token": "", "reason": "app-focus", "locale": self.current_language, "theme": "dark", "metadata": { "device": { "type": "phone", "uniqueId": unique_id}, "os": { "name": "android", "version": "14", "abis": ["arm64-v8a"], "host": "android"}, "app": { "platform": "android"}, "version": { "package": VYPN_PACKAGE, "binary": VYPN_VERSION, "js": VYPN_VERSION}}, "appFocusTime": 0, "playerActive": False, "playDuration": 0, "devMode": False, "hasAddon": True, "castConnected": False, "package": VYPN_PACKAGE, "version": VYPN_VERSION, "process": "app", "firstAppStart": current_timestamp - 86400000, "lastAppStart": current_timestamp, "ipLocation": None, "adblockEnabled": True, "migrationApplied": False, "migrationTargetInstalled": False, "proxy": { "supported": ["ss"], "engine": "Mu", "ssVersion": "2022", "enabled": False, "autoServer": True, "id": ""}, "iap": { "supported": False, "error": ""}} headers = { "user-agent": "okhttp/4.11.0", "accept": "application/json", "content-type": "application/json; charset=utf-8", "accept-encoding": "gzip", } # Try PING_URL (vavoo.tv) first, then PING_URL2 (vypn.net) urls = [PING_URL, PING_URL2] sig = None for url in urls: try: r = self.session.post( url, json=payload, headers=headers, timeout=15) if r.status_code == 200: data = r.json() sig = data.get("addonSig") or data.get("mhub") if sig: break else: print( "[AddonSig] No addonSig received from {}".format(url)) except Exception as e: print( "[AddonSig] Request to {} failed: {}".format( url, e)) continue if sig: client_ip = self._get_cached_external_ip() if client_ip: sig = _rewrite_addon_sig_ip(sig, client_ip) self.addon_sig_data["sig"] = sig self.addon_sig_data["ts"] = now print("[AddonSig] Token obtained successfully") return sig else: print("[AddonSig] Unable to obtain token from any URL") return None except Exception as e: print(" Error updating addonSig: " + str(e)) if self.addon_sig_data["sig"]: print(" Using old token") return self.addon_sig_data["sig"] return None def initialize_proxy(self): """Initialize the proxy by loading the catalog with fallback""" try: print(" Initializing...") # First, obtain a valid token sig = self.refresh_addon_sig_if_needed() if not sig: print( " Warning: Could not get a valid token, but continuing anyway") # We may continue with an old token or no token at all # Load the catalog print(" Attempting to load catalog...") all_channels = self.load_catalog(sig) if not all_channels or len(all_channels) == 0: print(" Warning: Catalog is empty or failed to load") # Create an empty list but mark as initialized self.all_filtered_items = [] self.channels_by_id = {} self.channels_by_country = {} self.countries_list = [] self.initialized = True print(" Initialized with empty catalog") return True self.all_filtered_items = all_channels self.channels_by_id = {} self.channels_by_country = {} countries = set() for channel in all_channels: channel_id = channel.get("id") if channel_id: self.channels_by_id[channel_id] = channel country = channel.get("country") if country: self.channels_by_country.setdefault( country, []).append(channel) if country and country != "default": countries.add(country) self.countries_list = sorted(list(countries)) self.local_ip = self.get_local_ip(force_refresh=True) self.initialized = True # Analysis print( " ✓ Initialized: %d channels, %d countries" % (len(all_channels), len(countries)) ) return True except Exception as e: print(" Initialization error: %s" % str(e)) # Even in case of error, try to continue with an empty catalog print(" Continuing with empty catalog") self.all_filtered_items = [] self.channels_by_id = {} self.channels_by_country = {} self.countries_list = [] self.initialized = True return True # Always return True; the proxy can work even without a catalog def load_catalog(self, sig): """Load the complete catalog with improved error handling including JSON decode errors.""" try: catalog_headers = { "content-type": "application/json; charset=utf-8", "mediahubmx-signature": sig, "user-agent": "MediaHubMX/2", "accept": "*/*", "Accept-Language": self.current_language, "Accept-Encoding": "gzip, deflate", # Connection: close is intentional for short-lived catalog # requests } all_channels = [] cursor = None page = 1 max_retries = 3 # Safety cap: real catalogs run to a few thousand channels / # a few dozen pages (300 items/page). Nothing should bound # this loop otherwise - a buggy or malicious upstream # response that kept returning a non-empty cursor forever # would keep it running indefinitely, since each page fetch # is a real network call that only stops the loop via an # empty items list/cursor, not a page count. max_pages = 200 print("Loading catalog...") while page <= max_pages: catalog_payload = { "language": self.current_language, "region": self.current_region, "catalogId": "iptv", "id": "iptv", "adult": False, "search": "", "sort": "", "filter": {}, "cursor": cursor, "clientVersion": "3.0.2" } success = False last_exception = None for attempt in range(max_retries): try: print( "Fetching catalog page {0} (attempt {1}/{2})".format(page, attempt + 1, max_retries)) r_catalog = self.session.post( self.catalog_url, json=catalog_payload, headers=catalog_headers, timeout=90, verify=False ) # HTTP 451 -> immediately try next mirror if r_catalog.status_code == 451: self._switch_to_next_base("(HTTP 451 on catalog)") if attempt < max_retries - 1: continue # Last attempt on new mirror try: r_catalog = self.session.post( self.catalog_url, json=catalog_payload, headers=catalog_headers, timeout=30 ) except Exception as e: print("Mirror fallback also failed: " + str(e)) break if r_catalog.status_code == 502: print( "502 Bad Gateway on page {0}, attempt {1}".format( page, attempt + 1)) if attempt < max_retries - 1: select.select([], [], [], 2 ** attempt) continue else: print( "Giving up on page {0} after {1} attempts".format( page, max_retries)) break r_catalog.raise_for_status() # ---------- JSON decode handling ---------- try: catalog_data = r_catalog.json() except ValueError as json_err: print( "JSON decode error on page {0}, attempt {1}: {2}".format( page, attempt + 1, json_err)) if attempt == max_retries - 1: # last attempt -> switch base and retry page self._switch_to_next_base( "(JSON decode error)") break else: select.select([], [], [], 2 ** attempt) continue # ------------------------------------------ success = True break except requests.exceptions.HTTPError as e: last_exception = e print("HTTP error on page {0}: {1}".format(page, e)) if e.response is not None and e.response.status_code == 451: self._switch_to_next_base( "(HTTP 451 on catalog HTTPError)") if attempt < max_retries - 1: continue if e.response is not None and e.response.status_code == 502 and attempt < max_retries - 1: select.select([], [], [], 2 ** attempt) continue else: break except Exception as e: last_exception = e print("Error on page {0}: {1}".format(page, e)) if attempt < max_retries - 1: select.select([], [], [], 2 ** attempt) continue else: break if not success: print( "Failed to load page {0}, stopping catalog download".format(page)) if last_exception: print("Last error: {0}".format(last_exception)) break items = catalog_data.get("items", []) if not items: print("No more items on page {0}".format(page)) break items_processed = 0 for item in items: if item.get("type") == "iptv": group = item.get("group", "") base_country = group separators = ["➾", "⟾", "->", "→", "»", "›"] for sep in separators: if sep in base_country: base_country = base_country.split(sep)[ 0].strip() break if not base_country: base_country = "default" channel_data = { "country": base_country, "id": item["ids"]["id"], "name": item["name"], "url": item["url"], "logo": item.get("logo", ""), "group": group } all_channels.append(channel_data) items_processed += 1 print( "Page {0}: processed {1} items, total {2} channels".format( page, items_processed, len(all_channels))) cursor = catalog_data.get("nextCursor") if not cursor: print("No more pages, catalog complete") break page += 1 if page % 10 == 0: select.select([], [], [], 0.1) print( "Catalog loaded: {0} channels in {1} pages".format( len(all_channels), page - 1)) return all_channels except Exception as e: print("Catalog load error: %s" % str(e)) trace_error() if all_channels: print( "Returning {0} channels already loaded".format( len(all_channels))) return all_channels return None def resolve_with_retry( self, channel_url, max_retries=2, force_refresh=False): """Resolve URLs with short-lived caching and fast retries. force_refresh=True skips the cache read (still updates it with the fresh result below) - used when the caller already knows the cached URL is bad, e.g. a player EOF-retry, where reusing the same cached URL for up to resolve_cache_ttl would otherwise guarantee every retry fails the exact same way. """ if not channel_url: print(" No channel URL provided") return None now = time.time() if not force_refresh: with self.resolve_cache_lock: cached = self.resolve_cache.get(channel_url) if cached and (now - cached["ts"] < self.resolve_cache_ttl): return cached["url"] for attempt in range(max_retries): try: if attempt > 0: self.refresh_addon_sig_if_needed(force=True) resolve_headers = { "content-type": "application/json; charset=utf-8", "mediahubmx-signature": self.addon_sig_data["sig"], "user-agent": "MediaHubMX/2", "accept": "*/*", "Accept-Language": self.current_language, "Accept-Encoding": "gzip, deflate", "Connection": "close", } resolve_payload = { "language": self.current_language, "region": self.current_region, "url": channel_url, "clientVersion": "3.0.2" } print( " Resolving channel URL (attempt %d/%d)" % (attempt + 1, max_retries)) r_resolve = self.session.post( self.resolve_url, json=resolve_payload, headers=resolve_headers, timeout=30 ) if r_resolve.status_code == 451: self._switch_to_next_base("(HTTP 451 on resolve)") if attempt < max_retries - 1: continue if r_resolve.status_code == 502 and attempt < max_retries - 1: select.select([], [], [], 0.25) continue r_resolve.raise_for_status() result = decode_response(r_resolve) stream_url = None if isinstance(result, list) and result: stream_url = result[0].get("url") elif isinstance(result, dict): stream_url = result.get("url") or result.get("streamUrl") if stream_url: # Pop before inserting (rather than # OrderedDict.move_to_end(), which doesn't exist on # Python 2.7's collections.OrderedDict - it raised # AttributeError there, caught by the except below, # meaning every resolve attempt appeared to fail on # Python 2 even though the stream URL was already # fetched) - re-resolving an already-cached URL # updates its value, and a plain re-assignment # doesn't move an existing key's position, so without # this, eviction below would go by original insertion # order rather than last use, and could drop a # channel that's actively being refreshed while # keeping one nobody has touched in a while. with self.resolve_cache_lock: self.resolve_cache.pop(channel_url, None) self.resolve_cache[channel_url] = { "url": stream_url, "ts": time.time()} # Evict oldest 500 entries (OrderedDict preserves # insertion order in Py2+3) if len(self.resolve_cache) > 1000: keys = list(self.resolve_cache.keys())[:-500] for key in keys: self.resolve_cache.pop(key, None) print(" Successfully resolved channel URL") return stream_url print(" Resolve response missing URL") except requests.exceptions.HTTPError as e: print( " HTTP error in resolve attempt %d: %s" % (attempt + 1, str(e))) try: if e.response is not None and e.response.status_code == 451: self._switch_to_next_base( "(HTTP 451 on resolve HTTPError)") except Exception as e2: debug("_switch_to_next_base failed: {}".format(e2)) if attempt < max_retries - 1: select.select([], [], [], 0.25) except Exception as e: print( " Error in resolve attempt %d: %s" % (attempt + 1, str(e))) if attempt < max_retries - 1: select.select([], [], [], 0.25) print(" Failed to resolve channel URL after all retries") return None def get_local_ip(self, force_refresh=False): # Deliberately always 127.0.0.1, not the box's real LAN IP: the # /channels endpoint embeds this directly into the proxy URLs # returned for bouquet export (http://:PORT/vavoo?...), # and localhost is "the core trick that makes streams stable" # (see CLAUDE.md) - a real LAN IP would still often work, but # loses that guarantee for no benefit. The API docs' own example # showing a LAN-looking address is just misleading, not a # promise this should actually return one. return PROXY_HOST def stop(self): """Stop background workers/timers and close session (safe for Py2/3).""" try: self._stop_event.set() except Exception: pass try: if self.refresh_timer: self.refresh_timer.cancel() except Exception: pass try: self.session.close() except Exception: pass class VavooHTTPHandler(BaseHTTPRequestHandler): timeout = 10 def do_HEAD(self): # BaseHTTPRequestHandler returns 501 for any method without a # do_* handler - some EPG importers (and other HTTP clients) # probe a URL with HEAD before actually fetching it, and would # treat that 501 as the source being unreachable, silently # skipping it entirely rather than falling back to GET. Reuse # do_GET's routing/headers for every endpoint, just without a # body, which is what HEAD is supposed to return anyway. self._head_only = True try: self.do_GET() finally: self._head_only = False def safe_write(self, data): if getattr(self, '_head_only', False): return True try: if isinstance(data, unicode): data = data.encode('utf-8') elif not isinstance(data, bytes): data = str(data).encode('utf-8') self.wfile.write(data) except (socket.error, IOError): return False return True def safe_send_response(self, code, message=None): """Safe response sending""" try: if message: self.send_response(code, message) else: self.send_response(code) except (BrokenPipeError, ConnectionResetError): print("Client disconnected during response - ignoring") return False return True def do_GET(self): # client_address = self.client_address[0] # print(" Request {1} from {0}" # .format(client_address, self.path)) try: parsed_path = urlparse(self.path) query_params = parse_qs(parsed_path.query) if parsed_path.path == '/vavoo': channel_id = query_params.get('channel', [None])[0] if not channel_id: self.send_error(400, "Missing channel parameter") return channel = proxy.channels_by_id.get(channel_id) if hasattr( proxy, 'channels_by_id') else None if not channel: self.send_error(404, "Channel not found") return try: # 1. Resolve the Vavoo stream URL. force=1 (sent by # the player's EOF-retry path) bypasses the resolve # cache, since retrying against the same cached URL # that just failed would otherwise guarantee every # retry fails the same way. force_refresh = query_params.get( 'force', ['0'])[0] == '1' stream_url = proxy.resolve_with_retry( channel.get("url"), force_refresh=force_refresh) if not stream_url: self.send_error(404, "Stream not resolved") return # Redirect 302 all'URL dello stream self.send_response(302) self.send_header('Location', stream_url) self.end_headers() print("[Proxy] Redirect to: " + stream_url[:100]) except Exception as e: print("[Proxy] /vavoo error: " + str(e)) self.send_error(500, "Internal proxy error") elif parsed_path.path == '/stream': # Get the 'ref' parameter (the converted service reference) ref_encoded = query_params.get('ref', [None])[0] if not ref_encoded: self.send_error(400, "Missing ref parameter") return # Decode the ref (replace %3a back to :) ref = unquote(ref_encoded) # Load the sref map try: with open(SREF_MAP_FILE, 'r') as f: sref_map = load(f) except BaseException: sref_map = {} channel_id = sref_map.get(ref) if not channel_id: self.send_error(404, "Channel not found in map") return # Now get the channel object from proxy.channels_by_id channel = proxy.channels_by_id.get(channel_id) if hasattr( proxy, 'channels_by_id') else None if not channel: self.send_error(404, "Channel not found") return try: # 1. Get stream URL stream_url = proxy.resolve_with_retry(channel["url"]) if not stream_url: self.send_error(404, "Stream not resolved") return # 2-4 all live under one try/finally so upstream is # always closed, including if raise_for_status() raises # or the client disconnects before headers are sent - # previously the connection/response was only closed if # the forwarding loop below was actually reached. upstream = None counted_stream = False try: # 2. Connect to upstream with streaming (timeout # aumentato) upstream = proxy.session.get( stream_url, stream=True, timeout=( 10, 90)) # (connect, read) 10s/90s upstream.raise_for_status() # 3. Send headers to player if not self.safe_send_response(200): return self.send_header( 'Content-Type', upstream.headers.get( 'Content-Type', 'video/mp2t')) self.send_header('Connection', 'keep-alive') # Add headers to prevent caching self.send_header( 'Cache-Control', 'no-cache, no-store') self.end_headers() # HEAD requests get headers only - the caller # (do_HEAD) wants to know the stream is reachable # without paying for the actual video transfer. if getattr(self, '_head_only', False): return # 4. Forward data with timeout monitoring (chunk # size increased) proxy.stream_started() counted_stream = True last_data_time = time.time() try: # Increase chunk size from 65536 to 262144 (256KB) for chunk in upstream.iter_content( chunk_size=262144): if chunk: self.wfile.write(chunk) self.wfile.flush() last_data_time = time.time() else: if time.time() - last_data_time > 15: print( "[Proxy Stream] Upstream stalled for: " + channel_id) break select.select([], [], [], 0.1) except (socket.timeout, ConnectionError, BrokenPipeError) as e: print("[Proxy Stream] Network error: " + str(e)) except Exception as e: print("[Proxy Stream] Unexpected error: " + str(e)) finally: if upstream is not None: try: upstream.close() except Exception: pass if counted_stream: proxy.stream_ended() print( "[Proxy Stream] Finished for channel: " + channel_id) except Exception as e: print("[Proxy Stream] Error: " + str(e)) self.send_error(500, "Streaming error") elif parsed_path.path == '/channels': country = query_params.get('country', [None])[0] if not country: self.send_error(400, "Missing country parameter") return matching_channels = [] if hasattr(proxy, 'all_filtered_items'): for channel in proxy.all_filtered_items: channel_country = channel.get("country", "") if channel_country.lower() == country.lower(): matching_channels.append(channel) response_channels = [] local_ip = proxy.get_local_ip() for channel in matching_channels: channel_id = channel.get("id", "") if channel_id: proxy_url = "http://%s:%d/vavoo?channel=%s" % ( local_ip, PORT, channel_id) response_channels.append({ "id": channel_id, "name": channel.get("name", ""), "url": proxy_url, "logo": channel.get("logo", ""), "country": channel.get("country", country) }) if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(response_channels)): return elif parsed_path.path == '/catalog': if hasattr(proxy, 'all_filtered_items'): if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(proxy.all_filtered_items)): return else: self.send_error(404, "No catalog loaded") elif parsed_path.path == '/countries': countries = set() if hasattr(proxy, 'all_filtered_items'): for channel in proxy.all_filtered_items: country = channel.get("country", "") if country and country != "default": countries.add(country) countries_list = sorted(list(countries)) if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(countries_list)): return elif parsed_path.path == '/status': status = { "initialized": proxy.initialized, "channels_count": len( proxy.all_filtered_items), "addon_sig_valid": proxy.addon_sig_data["sig"] is not None, "addon_sig_age": int( time.time() - proxy.addon_sig_data["ts"]), "local_ip": proxy.get_local_ip(), "port": PORT} if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(status)): return # Redirect per-country EPG requests to GitHub raw files elif parsed_path.path.startswith('/epg/') and parsed_path.path.endswith('.xml'): # Lowercase: the actual files on GitHub are lowercase # (epg_it.xml, epg_fr.xml, ...) and raw.githubusercontent.com # paths are case-sensitive, unlike every other caller of a # country code in this codebase (_get_epg_feed_index(), # update_epg_sources(), etc.), which already lowercase # first - an uppercase/mixed-case request here would # otherwise redirect straight to a 404. country_code = parsed_path.path.split( '/')[-1].replace('.xml', '').lower() github_url = "{}/vavoo-player/master/epg_{}.xml".format( HOST_GIT, country_code) self.send_response(302) self.send_header('Location', github_url) self.end_headers() elif parsed_path.path == '/health': """Health check endpoint with detailed status""" try: now = time.time() token_age = now - proxy.addon_sig_data["ts"] token_valid = proxy.addon_sig_data["sig"] is not None needs_refresh = token_age > 300 # 8 minutes # Calculate token expiration ttl = max(0, TOKEN_ADDON_SIG - int(token_age)) # Check if proxy is initialized initialized = proxy.initialized channels_count = len( proxy.all_filtered_items) if initialized else 0 # Proxy status proxy_status = { "status": "healthy" if initialized and token_valid else "unhealthy", "initialized": initialized, "channels_count": channels_count, "token": { "valid": token_valid, "age": int(token_age), "ttl": ttl, "needs_refresh": needs_refresh, "expires_in": str(ttl) + "s" }, "system": { "uptime": int(now - proxy.start_time if hasattr(proxy, 'start_time') else 0), "heartbeat": int(now - proxy.last_heartbeat), "port": PORT, "local_ip": proxy.get_local_ip() }, "timestamp": now, "message": "Proxy is running normally" if initialized else "Proxy not initialized" } # Read-only health endpoint: no forced refresh here # self.send_response(200) if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.send_header( 'Cache-Control', 'no-cache, no-store, must-revalidate') self.send_header('Pragma', 'no-cache') self.send_header('Expires', '0') self.end_headers() if not self.safe_write(dumps(proxy_status)): return except Exception as e: error_response = { "status": "error", "message": str(e), "timestamp": time.time() } if not self.safe_send_response(500): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(error_response)): return elif parsed_path.path == '/refresh_token': sig = proxy.refresh_addon_sig_if_needed(force=True) response = { "status": "success" if sig else "error", "message": "Token refreshed" if sig else "Failed to refresh token"} if not self.safe_send_response(200): return self.send_header('Content-Type', 'application/json') self.end_headers() if not self.safe_write(dumps(response)): return elif parsed_path.path == '/shutdown': # Signal global stop to prevent the restart loop from # re-spawning the server STOP_EVENT.set() if not self.safe_send_response(200): return self.send_header('Content-Type', 'text/plain') self.end_headers() if not self.safe_write(b"Proxy shutting down..."): return def shutdown_server(): select.select([], [], [], 0.2) if proxy.server: try: proxy.server.shutdown() except Exception: pass try: proxy.server.server_close() except Exception: pass try: proxy.stop() except Exception: pass t = threading.Thread(target=shutdown_server) t.setDaemon(True) # Py2/3 compatible t.start() else: self.send_error(404, "Not Found") except BrokenPipeError: print(" Client disconnected (BrokenPipeError)") return except ConnectionResetError: print(" Connection reset by client") return except Exception as e: print("[Handler] Error: %s" % str(e)) try: self.send_error(500, "Internal Server Error") except (BrokenPipeError, ConnectionResetError): print(" Client gone while sending error") return def handle_one_request(self): """Handle a single request with guaranteed cleanup""" try: BaseHTTPRequestHandler.handle_one_request(self) except (socket.timeout, socket.error) as e: print(" Socket error in request: " + str(e)) try: self.connection.close() except BaseException: pass except Exception as e: print(" Unexpected error in request: " + str(e)) def finish(self): """Override finish to manage cleanup""" try: BaseHTTPRequestHandler.finish(self) except (BrokenPipeError, ConnectionResetError): pass def setup(self): """Setup with timeout""" BaseHTTPRequestHandler.setup(self) self.request.settimeout(self.timeout) # Streaming writes are flushed per-chunk (see the /stream handler); # without this, Nagle's algorithm can hold small/odd-sized writes # briefly waiting to coalesce them, adding needless latency. try: self.request.setsockopt( socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) except (socket.error, OSError, AttributeError): pass def log_message(self, format, *args): pass proxy = VavooProxy() def shutdown_proxy(): """Shutdown the proxy server if running. Sets STOP_EVENT first so background threads stop even if HTTP call fails. """ # Signal all proxy threads to exit regardless of HTTP outcome STOP_EVENT.set() try: response = requests.get( PROXY_SHUTDOWN_URL, timeout=2) if response.status_code == 200: print(" Shutdown request sent successfully") STOP_EVENT.clear() # reset for next start return True except Exception as e: print(" Shutdown via HTTP failed: {}".format(e)) # Fallback: kill process try: import subprocess subprocess.call(["pkill", "-f", "python.*vavoo_proxy"]) print(" Killed by pkill") STOP_EVENT.clear() return True except Exception as e: print(" Failed to kill process: {}".format(e)) return False def start_proxy(): """Start the proxy server with restart on failure""" global proxy import subprocess # IMPORTANT: allow restart only if the current process is not running # (the PID file may be stale if the process has died) if is_proxy_running(): print("[PROXY] Proxy is already running, checking if it's responsive...") try: resp = requests.get(PROXY_STATUS_URL, timeout=2) if resp.status_code == 200: print("[PROXY] Proxy is responsive, not starting another instance") return True else: print("[PROXY] Proxy is running but not responsive, will restart") except Exception: print("[PROXY] Proxy running but not responding, restarting...") # Kill the unresponsive process try: subprocess.call(["pkill", "-f", "python.*vavoo_proxy"]) select.select([], [], [], 2) except Exception as e: debug("pkill of unresponsive proxy failed: {}".format(e)) # Write the PID file for this instance write_pid_file() # Indicate that we are booting write_booting_file() STOP_EVENT.clear() max_restarts = 3 restart_count = 0 while restart_count < max_restarts: try: print("=" * 50) print("VAVOO PROXY v" + str(__version__) + " (Attempt " + str(restart_count + 1) + "/" + str(max_restarts) + ")") print("=" * 50) if not proxy.initialize_proxy(): print("[✗] Failed to initialize proxy") restart_count += 1 if restart_count < max_restarts: select.select([], [], [], 3) proxy = VavooProxy() # Recreate proxy continue else: print("[✗] Max restart attempts reached") return False server = ThreadedHTTPServer(('0.0.0.0', PORT), VavooHTTPHandler) # Boot completed, remove booting file remove_booting_file() server.timeout = 30 server.request_queue_size = 64 proxy.server = server local_ip = proxy.get_local_ip() print("[✓] Channels: " + str(len(proxy.all_filtered_items))) print("[✓] IP: " + str(local_ip) + ":" + str(PORT)) print("[✓] Timeout: " + str(server.timeout) + "s") print("[✓] Ready") print("=" * 50) # Reset restart counter on success restart_count = 0 try: server.serve_forever(poll_interval=0.5) # If serve_forever returned, decide whether to restart or exit if STOP_EVENT.is_set(): print("[!] Shutdown requested, exiting proxy loop") try: server.server_close() except Exception: pass try: proxy.stop() except Exception: pass break except KeyboardInterrupt: print("\n[!] Proxy stopped by user") break except Exception as e: print("[✗] Server error: " + str(e)) restart_count += 1 if restart_count < max_restarts: print("[!] Restarting proxy in 5 seconds...") select.select([], [], [], 5) # Shutdown old server if exists if proxy.server: try: proxy.server.shutdown() proxy.server.server_close() try: proxy.stop() except Exception: pass except BaseException as e: debug("Old proxy server cleanup before " "restart failed: {}".format(e)) proxy = VavooProxy() # Recreate proxy continue except Exception as e: print("[✗] Critical error: " + str(e)) trace_error() restart_count += 1 if restart_count < max_restarts: print("[!] Restarting proxy in 5 seconds...") select.select([], [], [], 5) # Shutdown old server if exists if proxy.server: try: proxy.server.shutdown() proxy.server.server_close() except BaseException: pass proxy = VavooProxy() # Recreate proxy continue print("[✗] Proxy cannot start after " + str(max_restarts) + " attempts") return False def is_proxy_port_listening(): """Check if proxy is actually listening on PORT.""" import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) result = sock.connect_ex(('127.0.0.1', PORT)) sock.close() return result == 0 def run_proxy_in_background(startup_timeout=30): global _starting # Callers pass cfg.proxy_startup_timeout.value straight through, and # ConfigSelectionNumber.value has been confirmed (via a user log) to # come back as a str rather than an int on at least some # images/Python versions - min(str, int) raises TypeError on Python # 3, and even where it doesn't crash outright (Python 2's # cross-type comparison happens to usually favor the int here), it's # relying on undefined/fragile behavior rather than anything # guaranteed. Coerce once at the top so every caller is covered. try: startup_timeout = int(startup_timeout) except (TypeError, ValueError): startup_timeout = 30 # Wait (briefly) for another booting instance. This can run on the # caller's own thread - including the Enigma2 UI/reactor thread in some # call paths - so it is deliberately capped well below startup_timeout # (which may be configured up to 300s) to avoid freezing the UI. # Callers already poll for actual readiness asynchronously afterwards. if is_proxy_booting(): wait_seconds = min(startup_timeout, 10) print("[Proxy] Another proxy is booting, waiting up to {} seconds...".format( wait_seconds)) max_attempts = int(wait_seconds * 2) for attempt in range(max_attempts): if not is_proxy_booting() and is_proxy_port_listening(): print("[Proxy] Proxy boot completed, instance is running") return True select.select([], [], [], 0.5) print("[Proxy] Boot still in progress after short wait, returning (caller will poll readiness)") return True # Final check: if already active and listening, exit if is_proxy_running() and is_proxy_port_listening(): print("[Proxy] Already running and listening, skipping start") return True # Stale PID file? Clean it up if os.path.exists(PID_FILE) and not is_proxy_port_listening(): try: with open(PID_FILE) as f: pid = int(f.read().strip()) os.kill(pid, 0) # verify process exists os.kill(pid, 9) # kill if it exists but is not listening select.select([], [], [], 0.2) except Exception: pass try: os.unlink(PID_FILE) except Exception: pass # Acquire lock for atomic startup with _starting_lock: if _starting: print("[Proxy] Already starting, skipping...") return False _starting = True try: proxy_thread = threading.Thread(target=start_proxy) proxy_thread.setDaemon(True) proxy_thread.start() return True finally: with _starting_lock: _starting = False if __name__ == "__main__": start_proxy()