osint-dashboard/app/camera_hls.py

"""HLS playlist rewriter so the in-page player can play CORS-blocked feeds.

The browser talks only to /api/cameras/{id}/hls.m3u8 and /hlsseg. We fetch
the real playlist, rewrite every URI to our proxy, and remember the hosts
that appeared so /hlsseg cannot be used as an open proxy.
"""

from __future__ import annotations

import re
import time
from urllib.parse import quote, unquote, urljoin, urlparse

import httpx
from fastapi import HTTPException
from fastapi.responses import Response

from camera_scraper import is_public_url

_UA = {"User-Agent": "osint-dashboard-hls/1.0"}
# camera_id -> (expiry_epoch, allowed_hosts)
_ALLOWED: dict[str, tuple[float, set[str]]] = {}
_TTL = 600.0
_URI_ATTR = re.compile(r'URI="([^"]+)"', re.I)


def _remember(camera_id: str, url: str) -> None:
    host = urlparse(url).hostname
    if not host:
        return
    now = time.monotonic()
    exp, hosts = _ALLOWED.get(camera_id, (now + _TTL, set()))
    hosts.add(host.lower())
    _ALLOWED[camera_id] = (now + _TTL, hosts)


def _host_ok(camera_id: str, url: str) -> bool:
    host = (urlparse(url).hostname or "").lower()
    if not host:
        return False
    rec = _ALLOWED.get(camera_id)
    if not rec or rec[0] < time.monotonic():
        return False
    return host in rec[1]


def _proxied(camera_id: str, abs_url: str) -> str:
    _remember(camera_id, abs_url)
    return f"/api/cameras/{camera_id}/hlsseg?u={quote(abs_url, safe='')}"


def rewrite_m3u8(text: str, base: str, camera_id: str) -> str:
    out: list[str] = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            out.append(line)
            continue
        if stripped.startswith("#"):
            def repl(m: re.Match[str]) -> str:
                return f'URI="{_proxied(camera_id, urljoin(base, m.group(1)))}"'
            out.append(_URI_ATTR.sub(repl, line))
            continue
        out.append(_proxied(camera_id, urljoin(base, stripped)))
    return "\n".join(out) + "\n"


async def fetch_playlist(camera_id: str, url: str) -> Response:
    if not is_public_url(url):
        raise HTTPException(400, "HLS URL is not public")
    _remember(camera_id, url)
    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True, headers=_UA) as c:
            r = await c.get(url)
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(502, "HLS playlist unavailable") from exc
    if r.status_code != 200:
        raise HTTPException(502, "HLS playlist unavailable")
    base = str(r.url)
    body = rewrite_m3u8(r.text, base, camera_id)
    return Response(
        content=body,
        media_type="application/vnd.apple.mpegurl",
        headers={"Cache-Control": "no-store"},
    )


async def fetch_segment(camera_id: str, raw_url: str) -> Response:
    url = unquote(raw_url)
    if not url.lower().startswith(("http://", "https://")):
        raise HTTPException(400, "invalid segment URL")
    if not is_public_url(url):
        raise HTTPException(400, "segment URL is not public")
    if not _host_ok(camera_id, url):
        raise HTTPException(400, "segment host not in playlist")
    try:
        async with httpx.AsyncClient(timeout=20, follow_redirects=True, headers=_UA) as c:
            r = await c.get(url)
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(502, "HLS segment unavailable") from exc
    if r.status_code != 200:
        raise HTTPException(502, "HLS segment unavailable")
    ctype = (r.headers.get("content-type") or "").lower()
    if "mpegurl" in ctype or "m3u8" in url.lower().split("?")[0]:
        body = rewrite_m3u8(r.text, str(r.url), camera_id)
        return Response(content=body, media_type="application/vnd.apple.mpegurl",
                        headers={"Cache-Control": "no-store"})
    return Response(
        content=r.content,
        media_type=r.headers.get("content-type") or "application/octet-stream",
        headers={"Cache-Control": "no-store"},
    )