osint-dashboard/app/camera_preview.py

"""Resolve a browser-renderable preview for a camera.

HTTP/MJPEG cameras already expose a snapshot_url the existing proxy can
stream. masscan finds are stored as `rtsp://IP/` with no snapshot_url, so
the map popup used to skip the <img> entirely and the leftover source link
handed the browser an rtsp:// URL (which opens VLC).

This module:
  1. Tries a short list of unauthenticated HTTP snapshot paths (fast).
  2. Falls back to grabbing one JPEG frame from RTSP via ffmpeg (no auth).
  3. Remembers the first URL that worked on the camera row so the next
     popup is a cache hit.

No credentials are ever tried.
"""

from __future__ import annotations

import asyncio
import logging
import shutil
from urllib.parse import urlparse

import httpx

from camera_config import USER_AGENT
from camera_models import cameras
from camera_scraper import fetch_snapshot
from database import async_session

logger = logging.getLogger("osint.camera_preview")

# Most common unauthenticated still-image endpoints on consumer NVRs/IP cams.
# Keep this list SHORT — it runs on popup click.
_HTTP_PATHS = (
    "/snapshot.jpg",
    "/cgi-bin/snapshot.cgi",
    "/jpg/image.jpg",
    "/image.jpg",
    "/onvif/snapshot",
    "/axis-cgi/jpg/image.cgi",
    "/tmpfs/auto.jpg",
)

# Browser-playable MJPEG paths the /stream proxy can pass through.
_MJPEG_PATHS = (
    "/mjpg/video.mjpg",
    "/video.mjpg",
    "/cgi-bin/mjpg/video.cgi",
    "/axis-cgi/mjpg/video.cgi",
    "/nphMotionJpeg",
    "/mjpeg.cgi",
)

_FFMPEG = shutil.which("ffmpeg")


def _host_from_url(url: str) -> str | None:
    try:
        return urlparse(url).hostname
    except Exception:  # noqa: BLE001
        return None


def _looks_like_jpeg(data: bytes) -> bool:
    return bool(data) and len(data) >= 64 and data[:2] == b"\xff\xd8"


async def _http_get_image(url: str, timeout: float = 2.5) -> bytes | None:
    try:
        async with httpx.AsyncClient(
            timeout=timeout, follow_redirects=True,
            headers={"User-Agent": USER_AGENT},
        ) as c:
            r = await c.get(url)
        if r.status_code != 200:
            return None
        ctype = (r.headers.get("content-type") or "").lower()
        if "html" in ctype or "text/" in ctype:
            return None
        if not _looks_like_jpeg(r.content) and "image/" not in ctype:
            return None
        if len(r.content) < 64:
            return None
        return r.content
    except Exception:  # noqa: BLE001
        return None


async def _http_feed_url(url: str, timeout: float = 2.5) -> str | None:
    """Return url if it looks like an unauthenticated image/MJPEG feed."""
    try:
        async with httpx.AsyncClient(
            timeout=timeout, follow_redirects=True,
            headers={"User-Agent": USER_AGENT},
        ) as c:
            async with c.stream("GET", url) as r:
                if r.status_code != 200:
                    return None
                ctype = (r.headers.get("content-type") or "").lower()
                if "html" in ctype or ctype.startswith("text/"):
                    return None
                if any(x in ctype for x in ("image/", "multipart", "mjpeg", "octet-stream")):
                    # Read a little to reject empty/error bodies.
                    chunk = b""
                    async for b in r.aiter_bytes():
                        chunk += b
                        if len(chunk) >= 64:
                            break
                    if len(chunk) < 64:
                        return None
                    if b"html" in chunk[:64].lower():
                        return None
                    return url
    except Exception:  # noqa: BLE001
        return None
    return None


async def probe_public_feed(host: str) -> str | None:
    """Unauthenticated HTTP still or MJPEG URL for this host, or None.

    Used at masscan ingest time so dead RTSP-only hosts never hit the map.
    No credentials, no RTSP path-walking (too slow / rarely public).
    """
    urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
    urls.append(f"http://{host}:8080/shot.jpg")
    urls.extend(f"http://{host}{p}" for p in _MJPEG_PATHS)
    results = await asyncio.gather(
        *(_http_feed_url(u) for u in urls),
        return_exceptions=True,
    )
    for url, hit in zip(urls, results):
        if isinstance(hit, str) and hit:
            return hit
    return None


async def ffmpeg_snapshot(url: str, timeout: float = 8.0) -> bytes | None:
    """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails."""
    if not _FFMPEG or not url.lower().startswith("rtsp://"):
        return None
    cmd = [
        _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
        "-rtsp_transport", "tcp",
        "-timeout", "4000000",          # 4s socket timeout, microseconds
        "-i", url,
        "-frames:v", "1",
        "-f", "image2pipe", "-vcodec", "mjpeg",
        "pipe:1",
    ]
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
    except FileNotFoundError:
        return None
    try:
        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        try:
            await proc.wait()
        except Exception:  # noqa: BLE001
            pass
        return None
    if proc.returncode not in (0, None) or not _looks_like_jpeg(stdout or b""):
        return None
    return stdout


async def ffmpeg_mjpeg_stream(url: str):
    """Yield an MJPEG multipart body transcoded from RTSP. Caller streams it."""
    if not _FFMPEG:
        return
    cmd = [
        _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
    ]
    if url.lower().startswith("rtsp://"):
        cmd += ["-rtsp_transport", "tcp", "-timeout", "4000000"]
    cmd += [
        "-i", url,
        "-an", "-c:v", "mjpeg", "-q:v", "8",
        "-f", "mpjpeg", "pipe:1",
    ]
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.DEVNULL,
    )
    try:
        assert proc.stdout is not None
        while True:
            chunk = await proc.stdout.read(64 * 1024)
            if not chunk:
                break
            yield chunk
    finally:
        if proc.returncode is None:
            proc.kill()
            try:
                await proc.wait()
            except Exception:  # noqa: BLE001
                pass


async def _remember(camera_id, url: str) -> None:
    try:
        async with async_session() as session:
            await session.execute(
                cameras.update()
                .where(cameras.c.id == camera_id)
                .values(snapshot_url=url)
            )
            await session.commit()
    except Exception:  # noqa: BLE001
        logger.warning("failed to persist snapshot_url for %s", camera_id,
                       exc_info=True)


async def resolve_preview(row) -> tuple[bytes | None, str | None]:
    """Return (jpeg_bytes, working_url) for a cameras-table row.

    Tries, in order:
      * existing HTTP snapshot_url (via the TTL cache)
      * common HTTP still-image paths on the host
      * ffmpeg frame grab from the stored RTSP URL, then common RTSP paths
    """
    snap = row.get("snapshot_url") or ""
    source = row.get("source_url") or ""
    host = _host_from_url(snap) or _host_from_url(source)
    if not host:
        return None, None

    # 1. Known HTTP snapshot — go through the existing TTL cache.
    if snap.lower().startswith(("http://", "https://")):
        data = await fetch_snapshot(snap)
        if data:
            return data, snap

    # 2. Probe unauthenticated HTTP stills in parallel (fast fail) BEFORE
    # any ffmpeg — most open cams that preview at all do it over HTTP.
    http_urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
    http_urls.append(f"http://{host}:8080/shot.jpg")
    results = await asyncio.gather(
        *(_http_get_image(u) for u in http_urls),
        return_exceptions=True,
    )
    for url, data in zip(http_urls, results):
        if isinstance(data, (bytes, bytearray)) and data:
            await _remember(row["id"], url)
            return bytes(data), url

    # 3. One ffmpeg grab of the stored RTSP URL. Path-walking is too slow
    # for a popup click; unauthenticated RTSP often needs a vendor path
    # and/or credentials we will not try.
    rtsp_url = snap if snap.lower().startswith("rtsp://") else source
    if rtsp_url.lower().startswith("rtsp://"):
        data = await ffmpeg_snapshot(rtsp_url, timeout=5.0)
        if data:
            if not snap:
                await _remember(row["id"], rtsp_url)
            return data, rtsp_url

    return None, None