| import os |
| import json |
| import time |
| import secrets |
| from urllib.parse import urlparse, urljoin |
|
|
| import httpx |
| from fastapi import FastAPI, HTTPException, Header |
| from fastapi.responses import HTMLResponse, Response, JSONResponse |
| from pydantic import BaseModel |
|
|
| APP_NAME = "hf-hls-test-relay" |
| PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "").strip().rstrip("/") |
| RELAY_TOKEN = os.getenv("RELAY_TOKEN", "").strip() |
| ALLOWED_UPSTREAM_SUFFIXES = [ |
| x.strip().lower() |
| for x in os.getenv("ALLOWED_UPSTREAM_SUFFIXES", "").split(",") |
| if x.strip() |
| ] |
|
|
| |
| UPSTREAM_AUTHORIZATION = os.getenv("UPSTREAM_AUTHORIZATION", "").strip() |
|
|
| REGISTRY_PATH = "/tmp/hf_hls_test_registry.json" |
| app = FastAPI(title=APP_NAME) |
|
|
| class RegisterPayload(BaseModel): |
| key: str |
| url: str |
| headers: dict | None = None |
|
|
| def load_registry(): |
| try: |
| with open(REGISTRY_PATH, "r", encoding="utf-8") as f: |
| data = json.load(f) |
| return data if isinstance(data, dict) else {} |
| except Exception: |
| return {} |
|
|
| def save_registry(data): |
| tmp = REGISTRY_PATH + ".tmp" |
| with open(tmp, "w", encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, indent=2) |
| os.replace(tmp, REGISTRY_PATH) |
|
|
| def require_relay_token(authorization): |
| if not RELAY_TOKEN: |
| raise HTTPException(status_code=503, detail="RELAY_TOKEN is not configured") |
| expected = f"Bearer {RELAY_TOKEN}" |
| if not authorization or not secrets.compare_digest(authorization, expected): |
| raise HTTPException(status_code=401, detail="Unauthorized") |
|
|
| def host_allowed(url): |
| try: |
| parsed = urlparse(url) |
| except Exception: |
| return False |
| if parsed.scheme != "https" or not parsed.hostname: |
| return False |
| host = parsed.hostname.lower() |
| if not ALLOWED_UPSTREAM_SUFFIXES: |
| return False |
| return any(host == suffix or host.endswith("." + suffix) for suffix in ALLOWED_UPSTREAM_SUFFIXES) |
|
|
| def upstream_headers(custom_headers=None): |
| headers = { |
| "Accept": "*/*", |
| "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36", |
| } |
| if UPSTREAM_AUTHORIZATION: |
| headers["Authorization"] = UPSTREAM_AUTHORIZATION |
| if custom_headers: |
| headers.update(custom_headers) |
| return headers |
|
|
| async def fetch_upstream(url, custom_headers=None): |
| if not host_allowed(url): |
| host = urlparse(url).hostname or "" |
| raise HTTPException(status_code=400, detail=f"Upstream host is not allowed: {host}") |
| try: |
| async with httpx.AsyncClient(timeout=httpx.Timeout(20.0, connect=10.0), follow_redirects=True) as client: |
| r = await client.get(url, headers=upstream_headers(custom_headers)) |
| except httpx.TimeoutException: |
| raise HTTPException(status_code=504, detail="Upstream timeout") |
| except httpx.HTTPError as e: |
| raise HTTPException(status_code=502, detail=f"Upstream request failed: {type(e).__name__}") |
| return r |
|
|
| def rewrite_playlist(text, base_url, key): |
| registry = load_registry() |
| item = registry.get(key) |
| if not item: |
| raise HTTPException(status_code=404, detail="Unknown key") |
| children = item.setdefault("children", {}) |
|
|
| def token_for(abs_url): |
| for token, existing in children.items(): |
| if existing == abs_url: |
| return token |
| token = secrets.token_urlsafe(12) |
| children[token] = abs_url |
| return token |
|
|
| out = [] |
| for line in text.splitlines(): |
| stripped = line.strip() |
| if not stripped: |
| out.append(line) |
| continue |
| if stripped.startswith("#"): |
| marker = 'URI="' |
| idx = line.find(marker) |
| if idx >= 0: |
| start = idx + len(marker) |
| end = line.find('"', start) |
| if end > start: |
| raw_uri = line[start:end] |
| abs_url = urljoin(base_url, raw_uri) |
| token = token_for(abs_url) |
| line = line[:start] + f"/hls/{key}/{token}" + line[end:] |
| out.append(line) |
| continue |
| abs_url = urljoin(base_url, stripped) |
| token = token_for(abs_url) |
| out.append(f"/hls/{key}/{token}") |
|
|
| save_registry(registry) |
| return "\n".join(out) + "\n" |
|
|
| @app.get("/health") |
| async def health(): |
| return { |
| "ok": True, |
| "service": APP_NAME, |
| "streams": len(load_registry()), |
| "relay_token_configured": bool(RELAY_TOKEN), |
| "allowed_upstream_suffixes": ALLOWED_UPSTREAM_SUFFIXES, |
| "provider_auth_configured": bool(UPSTREAM_AUTHORIZATION), |
| } |
|
|
| @app.post("/api/register") |
| async def register(payload: RegisterPayload, authorization: str | None = Header(default=None)): |
| require_relay_token(authorization) |
| key = payload.key.strip() |
| url = payload.url.strip() |
| if not key: |
| raise HTTPException(status_code=400, detail="Missing key") |
| if not host_allowed(url): |
| host = urlparse(url).hostname or "" |
| raise HTTPException(status_code=400, detail=f"Upstream host is not allowed: {host}") |
|
|
| r = await fetch_upstream(url, payload.headers) |
| ctype = (r.headers.get("content-type") or "").lower() |
| body_start = r.content[:256].decode("utf-8", errors="ignore") |
|
|
| if r.status_code != 200: |
| return JSONResponse( |
| status_code=r.status_code, |
| content={ |
| "success": False, |
| "key": key, |
| "upstream_status": r.status_code, |
| "content_type": ctype, |
| "detail": "HF server could not access the upstream URL with the configured provider-approved credentials.", |
| }, |
| ) |
|
|
| if "#EXTM3U" not in body_start and "mpegurl" not in ctype: |
| raise HTTPException(status_code=400, detail=f"Upstream returned 200 but does not look like HLS. content-type={ctype}") |
|
|
| registry = load_registry() |
| registry[key] = {"url": url, "created_at": int(time.time()), "children": {}, "headers": payload.headers or {}} |
| save_registry(registry) |
|
|
| base = PUBLIC_BASE_URL or "" |
| return { |
| "success": True, |
| "key": key, |
| "type": "hls", |
| "player_url": f"{base}/player/{key}" if base else f"/player/{key}", |
| "media_url": f"{base}/media/{key}" if base else f"/media/{key}", |
| } |
|
|
| @app.get("/media/{key}") |
| async def media(key: str): |
| registry = load_registry() |
| item = registry.get(key) |
| if not item: |
| raise HTTPException(status_code=404, detail="Unknown key") |
| r = await fetch_upstream(item["url"], item.get("headers")) |
| if r.status_code != 200: |
| return Response(content=r.content, status_code=r.status_code, media_type=r.headers.get("content-type", "text/plain")) |
| ctype = (r.headers.get("content-type") or "").lower() |
| body_start = r.content[:64].decode("utf-8", errors="ignore").lstrip() |
| if "mpegurl" in ctype or body_start.startswith("#EXTM3U"): |
| rewritten = rewrite_playlist(r.text, str(r.url), key) |
| return Response(content=rewritten, status_code=200, media_type="application/vnd.apple.mpegurl", headers={"Cache-Control": "no-store"}) |
| return Response(content=r.content, status_code=200, media_type=r.headers.get("content-type", "application/octet-stream"), headers={"Cache-Control": "no-store"}) |
|
|
| @app.get("/hls/{key}/{token}") |
| async def hls_child(key: str, token: str): |
| registry = load_registry() |
| item = registry.get(key) |
| if not item: |
| raise HTTPException(status_code=404, detail="Unknown key") |
| url = (item.get("children") or {}).get(token) |
| if not url: |
| raise HTTPException(status_code=404, detail="Unknown child token") |
| r = await fetch_upstream(url, item.get("headers")) |
| if r.status_code != 200: |
| return Response(content=r.content, status_code=r.status_code, media_type=r.headers.get("content-type", "text/plain")) |
| ctype = (r.headers.get("content-type") or "").lower() |
| body_start = r.content[:64].decode("utf-8", errors="ignore").lstrip() |
| if "mpegurl" in ctype or body_start.startswith("#EXTM3U"): |
| rewritten = rewrite_playlist(r.text, str(r.url), key) |
| return Response(content=rewritten, status_code=200, media_type="application/vnd.apple.mpegurl", headers={"Cache-Control": "no-store"}) |
| return Response(content=r.content, status_code=200, media_type=r.headers.get("content-type", "application/octet-stream"), headers={"Cache-Control": "no-store"}) |
|
|
| @app.get("/player/{key}", response_class=HTMLResponse) |
| async def player(key: str): |
| if key not in load_registry(): |
| raise HTTPException(status_code=404, detail="Unknown key") |
| html = f'''<!doctype html> |
| <html><head> |
| <meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"> |
| <title>HF HLS Test Player</title> |
| <script src="https://cdn.jsdelivr.net/npm/hls.js@1"></script> |
| <style>html,body{{margin:0;background:#000;width:100%;height:100%;overflow:hidden}}video{{width:100%;height:100%;object-fit:contain;background:#000}}#msg{{position:absolute;inset:0;display:flex;align-items:center;justify-content:center;color:white;font-family:Arial,sans-serif}}</style> |
| </head><body> |
| <video id="video" controls playsinline autoplay></video><div id="msg">Loading stream…</div> |
| <script>(function(){{ |
| const video=document.getElementById("video"),msg=document.getElementById("msg"),src="/media/{key}"; |
| function ready(){{msg.style.display="none"}} function fail(t){{msg.textContent=t||"Stream failed"}} |
| if(video.canPlayType("application/vnd.apple.mpegurl")){{video.src=src;video.addEventListener("loadedmetadata",ready,{{once:true}});video.addEventListener("error",()=>fail("Native HLS failed"));}} |
| else if(window.Hls&&Hls.isSupported()){{const hls=new Hls({{enableWorker:false,lowLatencyMode:true,backBufferLength:30}});hls.loadSource(src);hls.attachMedia(video);hls.on(Hls.Events.MANIFEST_PARSED,ready);hls.on(Hls.Events.ERROR,function(_,data){{if(data&&data.fatal)fail("HLS failed: "+data.type);}});}} |
| else{{fail("HLS is not supported in this browser")}} |
| }})();</script> |
| </body></html>''' |
| return HTMLResponse(html, 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:;", |
| }) |
