import asyncio import hashlib import hmac import json import os import re import secrets import time from pathlib import Path from typing import Dict, Optional from urllib.parse import quote, urljoin, urlparse import httpx from fastapi import FastAPI, Header, HTTPException, Query, Request from fastapi.responses import HTMLResponse, Response, StreamingResponse from pydantic import BaseModel, Field APP_NAME = "player2-hf-relay" RELAY_TOKEN = os.getenv("RELAY_TOKEN", "").strip() SIGNING_SECRET = os.getenv("SIGNING_SECRET", "").strip() PUBLIC_BASE_URL = os.getenv( "PUBLIC_BASE_URL", "https://ezzygame-fot.hf.space", ).strip().rstrip("/") ALLOWED_UPSTREAM_SUFFIXES = [ x.strip().lower().lstrip(".") for x in os.getenv( "ALLOWED_UPSTREAM_SUFFIXES", "fastly.live.brightcove.com,tvsimo.net", ).split(",") if x.strip() ] REGISTRY_PATH = Path( os.getenv( "REGISTRY_PATH", "/tmp/player2_registry.json", ) ) EXTRACTOR_PATH = Path( os.getenv( "EXTRACTOR_PATH", "/app/extract-streams-v7-media.mjs", ) ) NODE_BIN = os.getenv("NODE_BIN", "node").strip() or "node" PUPPETEER_EXECUTABLE_PATH = os.getenv( "PUPPETEER_EXECUTABLE_PATH", "/usr/bin/chromium", ).strip() EXTRACT_TIMEOUT_SECONDS = max( 20, min( 120, int(os.getenv("EXTRACT_TIMEOUT_SECONDS", "55")), ), ) KEY_RE = re.compile( r"^[A-Za-z0-9._-]{1,128}$" ) ALLOWED_CAPTURED_HEADERS = { "accept", "accept-language", "origin", "referer", "user-agent", } FORWARDED_RESPONSE_HEADERS = { "accept-ranges", "content-length", "content-range", "content-type", "etag", "last-modified", } PLAYLIST_CONTENT_TYPES = ( "application/vnd.apple.mpegurl", "application/x-mpegurl", "audio/mpegurl", "audio/x-mpegurl", ) REQUEST_TIMEOUT = httpx.Timeout( connect=10.0, read=30.0, write=15.0, pool=10.0, ) MAX_REDIRECTS = 4 app = FastAPI( title=APP_NAME ) registry: Dict[str, dict] = {} registry_lock = asyncio.Lock() # Only one Chromium/V7 extraction is allowed at a time. extract_lock = asyncio.Lock() class RegisterRequest(BaseModel): key: str = Field( min_length=1, max_length=128, ) type: str url: str = Field( min_length=1, max_length=8000, ) headers: Dict[str, str] = Field( default_factory=dict ) class ExtractRegisterRequest(BaseModel): key: str = Field( min_length=1, max_length=128, ) watch_url: str = Field( min_length=1, max_length=8000, ) def fail( status: int, message: str, ): raise HTTPException( status_code=status, detail=message, ) def validate_key( key: str, ) -> str: key = key.strip() if not KEY_RE.fullmatch(key): fail( 400, "Invalid key", ) return key def normalize_type( value: str, ) -> str: value = value.strip().lower() if value not in { "hls", "mp4", }: fail( 400, "type must be hls or mp4", ) return value def hostname_allowed( hostname: Optional[str], ) -> bool: if not hostname: return False host = hostname.rstrip(".").lower() return any( host == suffix or host.endswith("." + suffix) for suffix in ALLOWED_UPSTREAM_SUFFIXES ) def validate_upstream_url( url: str, ) -> str: parsed = urlparse(url) if parsed.scheme != "https": fail( 400, "Only HTTPS upstream URLs are allowed", ) if parsed.username or parsed.password: fail( 400, "Credentials in upstream URL are not allowed", ) if not hostname_allowed(parsed.hostname): fail( 400, "Upstream host is not allowed: " + ( parsed.hostname or "unknown" ), ) return url def validate_watch_url( url: str, ) -> str: parsed = urlparse(url) if parsed.scheme != "https": fail( 400, "Only HTTPS watch URLs are allowed", ) if parsed.username or parsed.password: fail( 400, "Credentials in watch URL are not allowed", ) host = ( parsed.hostname or "" ).rstrip(".").lower() if not ( host == "tvsimo.net" or host.endswith(".tvsimo.net") ): fail( 400, "Only tvsimo.net watch URLs are allowed", ) return url def clean_header_value( value: str, ) -> str: return ( str(value) .replace("\r", "") .replace("\n", "") .strip()[:2000] ) def sanitize_headers( headers: Dict[str, str], ) -> Dict[str, str]: out = {} for key, value in ( headers or {} ).items(): key = ( str(key) .strip() .lower() ) if key not in ALLOWED_CAPTURED_HEADERS: continue value = clean_header_value( value ) if value: out[key] = value out.setdefault( "accept", "*/*", ) out.setdefault( "user-agent", ( "Mozilla/5.0 (X11; Linux x86_64) " "AppleWebKit/537.36 " "(KHTML, like Gecko) " "Chrome/151.0.0.0 " "Safari/537.36" ), ) return out def save_registry(): try: REGISTRY_PATH.parent.mkdir( parents=True, exist_ok=True, ) tmp = REGISTRY_PATH.with_suffix( ".tmp" ) tmp.write_text( json.dumps( registry, ensure_ascii=False, indent=2, ), encoding="utf-8", ) tmp.replace( REGISTRY_PATH ) except Exception as exc: print( f"[registry] save warning: {exc}" ) def load_registry(): global registry try: if not REGISTRY_PATH.exists(): return raw = json.loads( REGISTRY_PATH.read_text( encoding="utf-8" ) ) good = {} if isinstance( raw, dict, ): for key, item in raw.items(): try: key = validate_key( str(key) ) good[key] = { "type": normalize_type( str( item.get( "type", "", ) ) ), "url": validate_upstream_url( str( item.get( "url", "", ) ) ), "headers": sanitize_headers( item.get( "headers", {}, ) ), "updated_at": item.get( "updated_at" ), } except HTTPException: pass registry = good except Exception as exc: print( f"[registry] load warning: {exc}" ) def get_stream( key: str, ) -> dict: key = validate_key( key ) item = registry.get( key ) if not item: fail( 404, "Unknown stream key", ) return item def authorize( authorization: Optional[str], ): if not RELAY_TOKEN: fail( 503, "RELAY_TOKEN is not configured", ) if ( not authorization or not authorization.startswith( "Bearer " ) ): fail( 401, "Unauthorized", ) supplied = authorization[ 7: ].strip() if not secrets.compare_digest( supplied, RELAY_TOKEN, ): fail( 401, "Unauthorized", ) def sign( key: str, url: str, exp: int, ) -> str: if not SIGNING_SECRET: fail( 503, "SIGNING_SECRET is not configured", ) message = ( f"{key}\n{exp}\n{url}" ).encode() return hmac.new( SIGNING_SECRET.encode(), message, hashlib.sha256, ).hexdigest() def signed_fetch_url( key: str, url: str, ttl: int = 900, ) -> str: exp = ( int(time.time()) + ttl ) sig = sign( key, url, exp, ) return ( f"/hls/{quote(key, safe='')}/fetch" f"?url={quote(url, safe='')}" f"&exp={exp}" f"&sig={sig}" ) def verify_signature( key: str, url: str, exp: int, sig: str, ): now = int( time.time() ) if ( exp < now or exp > now + 3600 ): fail( 403, "Signed media URL expired or invalid", ) expected = sign( key, url, exp, ) if not secrets.compare_digest( expected, sig, ): fail( 403, "Invalid media signature", ) URI_ATTR_RE = re.compile( r'URI="([^"]+)"' ) def rewrite_playlist( body: bytes, base_url: str, key: str, ) -> bytes: text = body.decode( "utf-8", errors="replace", ) out = [] for raw in text.splitlines(): line = raw.strip() if not line: out.append( raw ) continue if line.startswith("#"): def repl(match): absolute = validate_upstream_url( urljoin( base_url, match.group(1), ) ) return ( f'URI="' f'{signed_fetch_url(key, absolute)}' f'"' ) out.append( URI_ATTR_RE.sub( repl, raw, ) ) else: absolute = validate_upstream_url( urljoin( base_url, line, ) ) out.append( signed_fetch_url( key, absolute, ) ) return ( "\n".join(out) + ( "\n" if text.endswith("\n") else "" ) ).encode() def is_playlist( url: str, content_type: str, body: bytes, ) -> bool: if ( urlparse(url) .path .lower() .endswith(".m3u8") ): return True ctype = ( content_type or "" ).lower() if any( value in ctype for value in PLAYLIST_CONTENT_TYPES ): return True return body.lstrip().startswith( b"#EXTM3U" ) async def fetch_bytes( url: str, headers: Dict[str, str], ): current = validate_upstream_url( url ) async with httpx.AsyncClient( timeout=REQUEST_TIMEOUT, follow_redirects=False, ) as client: for _ in range( MAX_REDIRECTS + 1 ): response = await client.get( current, headers=headers, ) if response.status_code in { 301, 302, 303, 307, 308, }: location = response.headers.get( "location" ) if not location: return ( response, response.content, current, ) current = validate_upstream_url( urljoin( current, location, ) ) continue return ( response, response.content, current, ) fail( 502, "Too many upstream redirects", ) async def open_stream( url: str, headers: Dict[str, str], ): current = validate_upstream_url( url ) client = httpx.AsyncClient( timeout=REQUEST_TIMEOUT, follow_redirects=False, ) try: for _ in range( MAX_REDIRECTS + 1 ): request = client.build_request( "GET", current, headers=headers, ) response = await client.send( request, stream=True, ) if response.status_code in { 301, 302, 303, 307, 308, }: location = response.headers.get( "location" ) await response.aclose() if not location: return ( client, response, current, ) current = validate_upstream_url( urljoin( current, location, ) ) continue return ( client, response, current, ) await client.aclose() fail( 502, "Too many upstream redirects", ) except Exception: await client.aclose() raise def passthrough_headers( response: httpx.Response, ) -> Dict[str, str]: out = {} for key, value in response.headers.items(): if ( key.lower() in FORWARDED_RESPONSE_HEADERS ): out[ key.lower() ] = value out[ "cache-control" ] = "no-store" return out def parse_extractor_result( stdout: str, ) -> dict: result_line = None for line in stdout.splitlines(): if line.startswith( "STREAM_RESULT_JSON=" ): result_line = line[ len("STREAM_RESULT_JSON="): ].strip() if not result_line: fail( 502, "V7 extractor returned no machine-readable result", ) try: results = json.loads( result_line ) except json.JSONDecodeError: fail( 502, "V7 extractor returned invalid JSON", ) if ( not isinstance(results, list) or not results or not isinstance(results[0], dict) ): fail( 502, "V7 extractor found no playable media", ) return results[0] async def run_v7_extractor( key: str, watch_url: str, ) -> dict: if not EXTRACTOR_PATH.is_file(): fail( 503, "V7 extractor file is missing", ) env = os.environ.copy() env[ "WATCH_URL" ] = watch_url env[ "WATCH_MATCH_ID" ] = key env[ "PUPPETEER_EXECUTABLE_PATH" ] = PUPPETEER_EXECUTABLE_PATH try: process = await asyncio.create_subprocess_exec( NODE_BIN, str(EXTRACTOR_PATH), cwd=str( EXTRACTOR_PATH.parent ), env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) except FileNotFoundError: fail( 503, "Node.js executable is unavailable", ) try: stdout_bytes, stderr_bytes = ( await asyncio.wait_for( process.communicate(), timeout=EXTRACT_TIMEOUT_SECONDS, ) ) except asyncio.TimeoutError: process.kill() await process.communicate() fail( 504, "V7 extraction timed out", ) stdout = stdout_bytes.decode( "utf-8", errors="replace", ) stderr = stderr_bytes.decode( "utf-8", errors="replace", ) # Keep useful diagnostics in HF logs, but never return the full # extraction log to public callers. if stdout: print( "[extractor stdout]\n" + stdout[-12000:] ) if stderr: print( "[extractor stderr]\n" + stderr[-12000:] ) if process.returncode != 0: fail( 502, f"V7 extractor failed with exit code {process.returncode}", ) return parse_extractor_result( stdout ) async def save_stream_record( key: str, media_type: str, media_url: str, headers: Dict[str, str], ) -> dict: key = validate_key( key ) media_type = normalize_type( media_type ) media_url = validate_upstream_url( media_url ) clean_headers = sanitize_headers( headers ) record = { "type": media_type, "url": media_url, "headers": clean_headers, "updated_at": int( time.time() ), } async with registry_lock: registry[ key ] = record save_registry() return record @app.on_event("startup") async def startup(): load_registry() print( "[startup] allowed:", ", ".join( ALLOWED_UPSTREAM_SUFFIXES ), ) print( "[startup] extractor:", str( EXTRACTOR_PATH ), ) print( "[startup] chromium:", PUPPETEER_EXECUTABLE_PATH, ) @app.get("/") async def root(): return { "ok": True, "service": APP_NAME, "streams": len( registry ), } @app.get("/health") async def health(): return { "ok": True, "service": APP_NAME, "streams": len( registry ), "relay_token_configured": bool( RELAY_TOKEN ), "signing_secret_configured": bool( SIGNING_SECRET ), "extractor_present": EXTRACTOR_PATH.is_file(), "extractor_busy": extract_lock.locked(), } @app.post("/api/register") async def register( payload: RegisterRequest, authorization: Optional[str] = Header( default=None ), ): authorize( authorization ) key = validate_key( payload.key ) record = await save_stream_record( key, payload.type, payload.url, payload.headers, ) return { "success": True, "key": key, "type": record[ "type" ], "player_url": f"{PUBLIC_BASE_URL}/player/{key}", "media_url": f"{PUBLIC_BASE_URL}/media/{key}", } @app.post("/api/extract-register") async def extract_register( payload: ExtractRegisterRequest, authorization: Optional[str] = Header( default=None ), ): authorize( authorization ) key = validate_key( payload.key ) watch_url = validate_watch_url( payload.watch_url.strip() ) async with extract_lock: result = await run_v7_extractor( key, watch_url, ) media = result.get( "media" ) if not isinstance( media, dict, ): fail( 502, "V7 result did not contain media", ) media_type = normalize_type( str( media.get( "type", "", ) ) ) selected_url = str( media.get( "url", "", ) ).strip() # For HLS, preserve the master playlist whenever V7 found one. if media_type == "hls": master_url = str( result.get( "masterUrl", "", ) or "" ).strip() if master_url: selected_url = master_url if not selected_url: fail( 502, "V7 result did not contain a media URL", ) record = await save_stream_record( key, media_type, selected_url, media.get( "headers", {}, ), ) return { "success": True, "key": key, "type": record[ "type" ], "source": result.get( "source" ), "player_url": f"{PUBLIC_BASE_URL}/player/{key}", "media_url": f"{PUBLIC_BASE_URL}/media/{key}", } @app.get("/api/status/{key}") async def status( key: str, ): item = get_stream( key ) return { "success": True, "key": key, "type": item[ "type" ], "updated_at": item.get( "updated_at" ), "player_url": f"{PUBLIC_BASE_URL}/player/{key}", "media_url": f"{PUBLIC_BASE_URL}/media/{key}", } @app.get("/media/{key}") async def media( key: str, request: Request, ): item = get_stream( key ) headers = dict( item[ "headers" ] ) if request.headers.get( "range" ): headers[ "range" ] = request.headers[ "range" ] if item[ "type" ] == "hls": try: response, body, final_url = ( await fetch_bytes( item[ "url" ], headers, ) ) except httpx.RequestError as exc: fail( 502, f"Upstream HLS request failed: {exc}", ) if response.status_code >= 400: return Response( body, status_code= response.status_code, media_type= response.headers.get( "content-type", "text/plain", ), ) return Response( rewrite_playlist( body, final_url, key, ), media_type= "application/vnd.apple.mpegurl", headers={ "cache-control": "no-store", }, ) try: client, response, _ = ( await open_stream( item[ "url" ], headers, ) ) except httpx.RequestError as exc: fail( 502, f"Upstream MP4 request failed: {exc}", ) async def iterator(): try: async for chunk in ( response.aiter_bytes() ): yield chunk finally: await response.aclose() await client.aclose() return StreamingResponse( iterator(), status_code= response.status_code, media_type= response.headers.get( "content-type", "video/mp4", ), headers= passthrough_headers( response ), ) @app.get("/hls/{key}/fetch") async def hls_fetch( key: str, request: Request, url: str = Query( ..., min_length=1, max_length=8000, ), exp: int = Query(...), sig: str = Query( ..., min_length=64, max_length=128, ), ): item = get_stream( key ) if item[ "type" ] != "hls": fail( 400, "Stream is not HLS", ) validate_upstream_url( url ) verify_signature( key, url, exp, sig, ) headers = dict( item[ "headers" ] ) if request.headers.get( "range" ): headers[ "range" ] = request.headers[ "range" ] if ( urlparse(url) .path .lower() .endswith(".m3u8") ): try: response, body, final_url = ( await fetch_bytes( url, headers, ) ) except httpx.RequestError as exc: fail( 502, f"Upstream HLS request failed: {exc}", ) if response.status_code >= 400: return Response( body, status_code= response.status_code, media_type= response.headers.get( "content-type", "text/plain", ), ) if is_playlist( final_url, response.headers.get( "content-type", "", ), body, ): return Response( rewrite_playlist( body, final_url, key, ), media_type= "application/vnd.apple.mpegurl", headers={ "cache-control": "no-store", }, ) return Response( body, status_code= response.status_code, media_type= response.headers.get( "content-type", "application/octet-stream", ), ) try: client, response, final_url = ( await open_stream( url, headers, ) ) except httpx.RequestError as exc: fail( 502, f"Upstream media request failed: {exc}", ) content_type = response.headers.get( "content-type", "", ) if any( value in content_type.lower() for value in PLAYLIST_CONTENT_TYPES ): try: body = await response.aread() finally: await response.aclose() await client.aclose() return Response( rewrite_playlist( body, final_url, key, ), media_type= "application/vnd.apple.mpegurl", headers={ "cache-control": "no-store", }, ) async def iterator(): try: async for chunk in ( response.aiter_bytes() ): yield chunk finally: await response.aclose() await client.aclose() return StreamingResponse( iterator(), status_code= response.status_code, media_type= content_type or "application/octet-stream", headers= passthrough_headers( response ), ) @app.get( "/player/{key}", response_class=HTMLResponse, ) async def player( key: str, ): item = get_stream( key ) media_type = json.dumps( item[ "type" ] ) media_url = json.dumps( f"/media/{key}" ) page = f""" Live Player
Loading stream…
""" return HTMLResponse( page, headers={ "cache-control": "no-store", "content-security-policy": ( "default-src 'self'; " "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; " "style-src 'self' 'unsafe-inline'; " "media-src 'self' blob:; " "connect-src 'self'; " "img-src 'self' data:; " "object-src 'none'; " "base-uri 'none'" ), }, )