bambuddy/backend/app/services/external_camera.py

"""External camera service.

Supports MJPEG streams, RTSP streams (via ffmpeg), HTTP snapshot URLs, and USB cameras.

Security Note: This service intentionally makes requests to user-configured camera URLs.
This is necessary functionality for external camera integration. URLs are validated
to ensure they are well-formed before use.
"""

import asyncio
import functools
import ipaddress
import logging
import re
import shutil
import socket
from collections.abc import AsyncGenerator, Callable
from pathlib import Path
from urllib.parse import urlparse

import aiohttp

from backend.app.core.logging_filters import redact_url_credentials
from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr

logger = logging.getLogger(__name__)

# Protocols ffmpeg may use for an RTSP input. RTSP negotiates its media
# transport at runtime, so the transports have to be here alongside rtsp itself;
# tls and crypto cover encrypted variants. Everything ffmpeg would otherwise
# accept behind an -i — file, http, tcp to anywhere, concat — is left out, so a
# stream that references something outside itself cannot pull it in.
_RTSP_PROTOCOL_WHITELIST = "rtsp,rtp,udp,tcp,tls,crypto"


def _blocked_host_reason(hostname: str) -> str | None:
    """Describe why *hostname* is a destination we refuse to fetch, or None to allow it.

    Camera URLs are user-supplied and reach the network — over aiohttp for the
    HTTP types, and as an ``ffmpeg -i`` argument for RTSP — so this is where the
    SSRF boundary sits. LAN addresses are deliberately allowed: cameras live on
    the same network as Bambuddy, and blocking RFC-1918 would remove the feature
    rather than protect it. What is left to refuse is the host talking to
    itself, the unspecified address, link-local (which is where the cloud
    metadata endpoint lives), and the metadata hostnames.

    IP literals are classified with ``ipaddress`` rather than compared against a
    list of spellings, because 127.0.0.1, 127.0.0.2, 2130706433, 0177.0.0.1,
    127.1 and ::ffff:127.0.0.1 all arrive at loopback and a list of strings only
    ever catches whichever one someone thought to write down. ``inet_aton``
    comes first because it accepts the legacy octal, decimal and short forms
    that ``ip_address`` rejects — the C resolvers behind aiohttp and ffmpeg
    accept them, so refusing to understand them here would only mean not seeing
    where the request is actually going.
    """
    host = hostname.lower()

    ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None = None
    try:
        ip = ipaddress.ip_address(socket.inet_aton(host))
    except OSError:
        try:
            ip = ipaddress.ip_address(host)
        except ValueError:
            ip = None

    if ip is None:
        # A name, not an address. It is not resolved here on purpose: aiohttp
        # and ffmpeg each resolve independently afterwards, so a check here
        # decides nothing about where they end up (DNS rebinding), while a
        # lookup on every capture would break LAN cameras behind slow or
        # intermittent local DNS.
        if host == "localhost" or host.endswith(".localhost"):
            return "localhost"
        if host in ("metadata.google.internal", "metadata.google"):
            return "a cloud metadata service"
        return None

    # ::ffff:127.0.0.1 is loopback wearing an IPv6 spelling.
    mapped = getattr(ip, "ipv4_mapped", None)
    if mapped is not None:
        ip = mapped

    if ip.is_loopback:
        return "loopback"
    if ip.is_unspecified:
        return "the unspecified address"
    if ip.is_link_local:
        return "a link-local address (the cloud metadata range)"
    return None


def _sanitize_camera_url(url: str, allowed_schemes: tuple[str, ...] = ("http", "https", "rtsp")) -> str | None:
    """Validate and sanitize camera URL, returning a safe reconstructed URL.

    This validates that the URL is well-formed, uses an allowed scheme, does not
    target the host itself or a cloud metadata service, and returns a URL
    reconstructed from the validated components.

    Note: This intentionally allows user-provided URLs as that is the
    purpose of external camera configuration. Local network IPs are
    allowed since cameras are typically on the same LAN.

    Args:
        url: URL to validate and sanitize
        allowed_schemes: Tuple of allowed URL schemes

    Returns:
        Sanitized URL string if valid, None otherwise
    """
    try:
        parsed = urlparse(url)
        if not parsed.scheme or not parsed.netloc:
            return None

        # Validate scheme against allowlist
        scheme = parsed.scheme.lower()
        if scheme not in allowed_schemes:
            return None

        hostname = parsed.hostname or ""
        if not hostname:
            return None
        blocked = _blocked_host_reason(hostname)
        if blocked:
            logger.warning("Blocked camera URL targeting %s: %s", blocked, hostname)
            return None

        # Reconstruct URL from validated components to break taint chain
        # This creates a new string from validated parts
        #
        # The credentials are carried across verbatim from netloc rather than
        # via parsed.username/.password, which urlparse has already percent-
        # decoded: re-emitting those would corrupt any password containing an
        # @ or a :. They have to survive at all because most RTSP cameras — and
        # a fair number of MJPEG ones — carry their login in the URL, and
        # dropping it turns every one of them into an authentication failure.
        netloc = parsed.netloc
        userinfo = f"{netloc.rsplit('@', 1)[0]}@" if "@" in netloc else ""
        # parsed.hostname has already stripped the brackets off an IPv6 literal;
        # without them back the result is not a URL any client can parse.
        host_str = f"[{hostname}]" if ":" in hostname else hostname
        port_str = f":{parsed.port}" if parsed.port else ""
        path = parsed.path or ""
        query = f"?{parsed.query}" if parsed.query else ""
        fragment = f"#{parsed.fragment}" if parsed.fragment else ""

        # Build sanitized URL from validated components
        sanitized = f"{scheme}://{userinfo}{host_str}{port_str}{path}{query}{fragment}"
        return sanitized
    except ValueError:
        return None


def _validate_camera_url(url: str, allowed_schemes: tuple[str, ...] = ("http", "https", "rtsp")) -> bool:
    """Validate camera URL format (legacy wrapper).

    Args:
        url: URL to validate
        allowed_schemes: Tuple of allowed URL schemes

    Returns:
        True if URL is valid, False otherwise
    """
    return _sanitize_camera_url(url, allowed_schemes) is not None


def list_usb_cameras() -> list[dict]:
    """List available USB cameras (V4L2 devices on Linux).

    Returns:
        List of dicts with {device: str, name: str, capabilities: list}
    """
    cameras = []
    video_devices = sorted(Path("/dev").glob("video*"))

    for device in video_devices:
        device_path = str(device)
        info = {"device": device_path, "name": device.name, "capabilities": []}

        # Try to get device info via v4l2-ctl
        v4l2_ctl = shutil.which("v4l2-ctl")
        if v4l2_ctl:
            import subprocess

            try:
                result = subprocess.run(
                    [v4l2_ctl, "-d", device_path, "--info"],
                    capture_output=True,
                    text=True,
                    timeout=5,
                )
                if result.returncode == 0:
                    # Parse device name from output
                    for line in result.stdout.splitlines():
                        if "Card type" in line:
                            info["name"] = line.split(":", 1)[1].strip()
                        elif "Driver name" in line:
                            info["driver"] = line.split(":", 1)[1].strip()

                    # Check if device supports video capture
                    result = subprocess.run(
                        [v4l2_ctl, "-d", device_path, "--list-formats"],
                        capture_output=True,
                        text=True,
                        timeout=5,
                    )
                    if result.returncode == 0 and result.stdout.strip():
                        info["capabilities"].append("capture")
                        # Parse available formats
                        formats = re.findall(r"'(\w+)'", result.stdout)
                        info["formats"] = list(set(formats))

            except (subprocess.TimeoutExpired, Exception) as e:
                logger.debug("v4l2-ctl failed for %s: %s", device_path, e)

        # Only include devices that look like video capture devices
        # Skip metadata devices (typically odd numbered like video1, video3)
        try:
            device_num = int(device.name.replace("video", ""))
            # Even numbered devices are usually capture, odd are metadata
            # But also check if we got capabilities
            if info.get("capabilities") or device_num % 2 == 0:
                cameras.append(info)
        except ValueError:
            cameras.append(info)

    return cameras


def get_ffmpeg_path() -> str | None:
    """Get the path to ffmpeg executable."""
    # Try shutil.which first
    path = shutil.which("ffmpeg")
    if path:
        return path
    # Check common locations (systemd services may have limited PATH)
    for common_path in ["/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg", "/opt/homebrew/bin/ffmpeg"]:
        if Path(common_path).exists():
            return common_path
    return None


# In-flight one-shot captures, keyed by (url, camera_type, snapshot_url) —
# the tuple that actually identifies the physical resource being contended
# (#2707 comment thread, following #2705's shape for the built-in path).
#
# V4L2 USB devices allow exactly one open handle, and is_stream_active() /
# try_get_active_buffered_frame() (#2707) only stop a one-shot capturer from
# competing with the fan-out live view. They do nothing for capturer-vs-
# capturer with no viewer attached, where every consumer correctly concludes
# it isn't competing with a viewer and then collides with the others -
# exactly the #2705 report, just for this module's callers instead of
# capture_camera_frame_bytes()'s (Obico polling, the in-print frame bank,
# the finish-photo moment, plate detection, and the notification snapshot
# all reach capture_frame() independently).
#
# snapshot_url is part of the key (not just url/camera_type) because it
# routes to a completely different endpoint (#1177) - two printers that
# share a camera_url but differ only in snapshot_url must not coalesce.
_inflight_captures: dict[tuple[str, str, str | None], asyncio.Task[bytes | None]] = {}


def capture_in_flight(url: str, camera_type: str, snapshot_url: str | None = None) -> bool:
    """Return True iff a one-shot capture for this key is running right now.

    Mirrors camera.py's capture_in_flight() for the built-in path - for a
    caller that needs to know it will JOIN someone else's capture rather
    than open its own connection. Ordinary consumers should ignore this:
    they want "a recent frame", and capture_frame() already does the right
    thing for them.
    """
    task = _inflight_captures.get((url, camera_type, snapshot_url))
    return task is not None and not task.done()


def _discard_inflight_capture(key: tuple[str, str, str | None], task: asyncio.Task) -> None:
    """Done-callback: drop the finished task from the in-flight registry.

    Guarded on identity so a slow task that finishes after a newer capture
    has registered for the same key can't evict its successor.

    Also retrieves the exception, if any: the leader normally awaits the
    task and would surface it, but a leader whose own caller was cancelled
    leaves nobody to collect it, and an unretrieved task exception is
    logged by asyncio as a warning with a traceback at an arbitrary later
    point otherwise.
    """
    if _inflight_captures.get(key) is task:
        del _inflight_captures[key]
    if not task.cancelled() and task.exception() is not None:
        logger.debug("In-flight external-camera capture for %s ended in an exception", _log_key(key))


def _log_key(key: tuple[str, str, str | None]) -> str:
    """Render an in-flight key for a log line, with credentials redacted.

    Unlike camera.py's coalescing — which is keyed by IP address and so has
    nothing to hide — these keys carry the camera URL, and an RTSP camera URL
    routinely embeds ``user:pass@``. Redact before truncating: slicing first
    can cut the URL short of the ``@`` the pattern anchors on and leave the
    password in the log, which is why every other URL log in this module does
    it in this order.
    """
    return redact_url_credentials(key[0])[:50] if key[0] else "None"


async def capture_frame(
    url: str,
    camera_type: str,
    timeout: int = 15,
    snapshot_url: str | None = None,
) -> bytes | None:
    """Capture single frame from external camera.

    Args:
        url: Live-stream URL (MJPEG stream, RTSP URL, HTTP snapshot URL, or USB device path).
        camera_type: "mjpeg", "rtsp", "snapshot", or "usb".
        timeout: Connection timeout in seconds. Applies to this caller's own
            wait, including when it joins another caller's capture - call
            sites disagree about the value, and a follower must not silently
            inherit the leader's deadline in either direction.
        snapshot_url: Optional override for single-frame capture. When set, fetched
            via plain HTTP GET regardless of `camera_type`. Bypasses MJPEG warm-up
            handling on sources that expose a dedicated frame endpoint (e.g. go2rtc's
            `/api/frame.jpeg` reliably returns a clean image while the MJPEG stream's
            first frame is often the encoder's stale keyframe). #1177.

    Returns:
        JPEG bytes or None on failure

    Concurrent callers for the same (url, camera_type, snapshot_url) share
    one capture (#2705-shape fix, filed for the external-camera path as a
    follow-up on #2707): the first opens the connection, everyone arriving
    while it's in flight awaits the same result. This coalesces; it does
    not cache - a call that arrives after the previous capture finished
    always captures fresh, since plate detection and the finish-photo path
    judge a running print from these frames and a stale one there is worse
    than a slow one (#1397).
    """
    key = (url, camera_type, snapshot_url)

    # A follower whose leader fails takes a turn of its own rather than
    # inheriting a failure it never had a chance to avoid - by then the
    # leader has finished, so there's no connection left to compete with.
    # Bounded at two rounds: if the capture we joined AND its replacement
    # both failed, a third attempt won't help, and this caller has already
    # spent its patience.
    for _ in range(2):
        leader = _inflight_captures.get(key)
        if leader is None or leader.done():
            break
        try:
            frame = await asyncio.wait_for(asyncio.shield(leader), timeout=timeout)
        except TimeoutError:
            # shield() keeps the capture running for whoever else is still
            # waiting on it - giving up is this caller's decision alone.
            logger.warning(
                "Gave up waiting %ss on the in-flight external-camera capture for %s", timeout, _log_key(key)
            )
            return None
        except asyncio.CancelledError:
            # Distinguish "the capture I joined was cancelled" from "I was
            # cancelled". Only the former is ours to recover from.
            if not leader.cancelled():
                raise
            logger.info("In-flight external-camera capture for %s was cancelled; capturing our own", _log_key(key))
            continue
        if frame is not None:
            logger.debug(
                "Reusing in-flight external-camera capture for %s: %d bytes (no second connection opened)",
                _log_key(key),
                len(frame),
            )
            return frame
        logger.debug("In-flight external-camera capture for %s failed; capturing our own", _log_key(key))
    else:
        return None

    task = asyncio.create_task(_capture_frame_uncoalesced(url, camera_type, timeout, snapshot_url))
    _inflight_captures[key] = task
    task.add_done_callback(functools.partial(_discard_inflight_capture, key))
    # No wait_for here: this caller IS the capture, and each dispatched
    # _capture_* function already enforces `timeout` internally, where it
    # can also kill the ffmpeg process - a second deadline on top would
    # abandon the subprocess instead of killing it. shield() so a cancelled
    # leader (a client navigating away mid-request is routine) doesn't take
    # the capture down with it - followers already waiting on it still get
    # their frame.
    return await asyncio.shield(task)


async def _capture_frame_uncoalesced(
    url: str,
    camera_type: str,
    timeout: int,
    snapshot_url: str | None,
) -> bytes | None:
    """Open a connection and capture one frame. See capture_frame().

    Callers want that wrapper, not this: it opens a connection
    unconditionally, which is the collision #2705/#2707 are about.

    Failure is reported as ``None``, never as an exception. That is load-
    bearing now that captures are shared: the coalescing wrapper hands one
    task's outcome to every caller waiting on it, and it can only give a
    follower its own turn for an outcome it can recognise. An exception
    escaping here would instead propagate to every follower at once —
    turning one caller's failure into N — and none of them would retry.
    The per-type helpers below each catch what they expect and return None,
    but they catch narrowly (``aiohttp.ClientError``/``OSError``/timeouts),
    so this is the structural guarantee rather than one contingent on their
    coverage. Mirrors ``_capture_camera_frame_bytes_uncoalesced`` in
    camera.py, which ends in the same blanket catch for the same reason.
    """
    try:
        if snapshot_url:
            # Redact before truncating — slicing first can cut the URL short of the
            # ``@`` the pattern anchors on and leave the password in the log.
            logger.debug("capture_frame using snapshot override url=%s...", redact_url_credentials(snapshot_url)[:50])
            return await _capture_snapshot(snapshot_url, timeout)
        logger.debug(
            "capture_frame called: type=%s, url=%s...",
            camera_type,
            redact_url_credentials(url)[:50] if url else "None",
        )
        if camera_type == "mjpeg":
            return await _capture_mjpeg_frame(url, timeout)
        elif camera_type == "rtsp":
            return await _capture_rtsp_frame(url, timeout)
        elif camera_type == "snapshot":
            return await _capture_snapshot(url, timeout)
        elif camera_type == "usb":
            return await _capture_usb_frame(url, timeout)
        else:
            logger.warning("Unknown camera type: %s", camera_type)
            return None
    except asyncio.CancelledError:
        # Cancellation is not a capture failure and must stay distinguishable:
        # the wrapper checks ``leader.cancelled()`` to decide whether a
        # follower may take its own turn.
        raise
    except Exception:
        logger.exception("External camera capture failed for %s", redact_url_credentials(url)[:50] if url else "None")
        return None


def _safe_usb_device_path(device: str) -> str | None:
    """Rebuild a /dev/videoN path from a validated device number, or None.

    Validate device path - must be /dev/videoN format where N is 0-99. This
    prevents path traversal by using a strict allowlist approach: the returned
    path is built from an integer, which cannot carry a traversal, rather than
    from any part of the caller's string.

    Returns None if the device does not exist, so a caller cannot hand ffmpeg a
    path to something that is not a device node.
    """
    device_match = re.match(r"^/dev/video(\d{1,2})$", device)
    if not device_match:
        logger.error("Invalid USB device path format: %s", device)
        return None

    # Convert to integer to break taint chain - integers cannot contain path traversal
    # lgtm[py/path-injection] - device_num is validated integer 0-99
    device_num = int(device_match.group(1))  # Safe: regex guarantees 1-2 digits

    # Construct safe path from validated integer (completely untainted)
    safe_device_path = Path(f"/dev/video{device_num}")  # lgtm[py/path-injection]

    if not safe_device_path.exists():
        logger.error("USB device does not exist: %s", safe_device_path)
        return None

    return str(safe_device_path)  # lgtm[py/path-injection]


async def _capture_usb_frame(device: str, timeout: int) -> bytes | None:
    """Capture frame from USB camera using ffmpeg."""
    ffmpeg = get_ffmpeg_path()
    if not ffmpeg:
        logger.error("ffmpeg not found - required for USB camera capture")
        return None

    safe_device = _safe_usb_device_path(device)
    if not safe_device:
        return None

    # Use the safe path for ffmpeg - this is a hardcoded /dev/videoN path
    device = safe_device  # lgtm[py/path-injection]

    # Use ffmpeg to grab a single frame from USB camera
    cmd = [
        ffmpeg,
        "-f",
        "v4l2",
        "-i",
        device,
        "-frames:v",
        "1",
        "-f",
        "image2pipe",
        "-vcodec",
        "mjpeg",
        "-q:v",
        "2",
        "-",
    ]

    try:
        logger.debug("Running USB capture: %s", " ".join(cmd))
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)

        if process.returncode != 0:
            logger.error("ffmpeg USB capture failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT)
            return None

        if not stdout or len(stdout) < 100:
            logger.error("ffmpeg returned empty or too small frame from USB camera")
            return None

        return stdout

    except TimeoutError:
        logger.warning("USB frame capture timed out after %ss", timeout)
        if process:
            process.kill()
        return None
    except OSError as e:
        logger.error("USB frame capture failed: %s", e)
        return None


async def _capture_mjpeg_frame(url: str, timeout: int) -> bytes | None:
    """Extract a single representative frame from an MJPEG stream.

    Many MJPEG sources — go2rtc most notably (#1177), and several IP cameras —
    emit a "warm-up" frame on the byte that follows connection accept: usually
    the last keyframe held in the encoder, which is often black or stale until
    the encoder catches up to live content. To return a frame that's actually
    representative of the scene we read past the first frame and return the
    second; if the connection closes / times out / hits the buffer cap before
    a second frame ever arrives we fall back to the first so callers still
    get *something* (better than degrading slow / single-frame streams to None,
    which would regress every code path that consumed pre-fix behaviour).

    Note: this function intentionally makes requests to user-configured URLs.
    External camera support requires connecting to user-specified camera
    endpoints. URL is sanitized and dangerous destinations are blocked.
    """
    safe_url = _sanitize_camera_url(url, ("http", "https"))
    if not safe_url:
        logger.error("Invalid MJPEG URL format: %s...", redact_url_credentials(url)[:50])
        return None

    jpeg_start = b"\xff\xd8"
    jpeg_end = b"\xff\xd9"
    first_frame: bytes | None = None  # warm-up frame; fallback if no second arrives
    buffer = b""

    try:
        async with (
            aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session,
            session.get(safe_url) as response,
        ):
            if response.status != 200:
                logger.error("MJPEG stream returned status %s", response.status)
                return None

            async for chunk in response.content.iter_chunked(8192):
                buffer += chunk

                # A single chunk can carry multiple frames (e.g. high-FPS sources)
                # or a partial frame. Drain every complete frame we already have
                # before pulling the next chunk.
                while True:
                    start_idx = buffer.find(jpeg_start)
                    if start_idx == -1:
                        # No frame start yet — drop trailing garbage, keep waiting.
                        break
                    end_idx = buffer.find(jpeg_end, start_idx + 2)
                    if end_idx == -1:
                        # Partial frame; trim already-discarded prefix so the
                        # buffer stays bounded across long-running streams.
                        if start_idx > 0:
                            buffer = buffer[start_idx:]
                        break
                    frame = buffer[start_idx : end_idx + 2]
                    buffer = buffer[end_idx + 2 :]
                    if first_frame is None:
                        first_frame = frame  # warm-up; keep but don't return yet
                        continue
                    return frame  # representative second frame

                if len(buffer) > 5 * 1024 * 1024:  # 5MB limit
                    logger.warning("MJPEG buffer exceeded 5MB without finding frame")
                    break  # exit chunk loop, fall through to first_frame fallback

    except TimeoutError:
        logger.warning("MJPEG frame capture timed out after %ss", timeout)
    except (aiohttp.ClientError, OSError) as e:
        logger.error("MJPEG frame capture failed: %s", e)

    # Stream ended / timed out / buffer cap before a second frame arrived.
    # Return whatever warm-up frame we managed to read; better an iffy frame
    # than None for callers that need *some* image (snapshot UX, plate-detect
    # CV, finish photo). None only if no frame ever arrived at all.
    return first_frame


async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None:
    """Capture frame from RTSP using ffmpeg.

    For rtsps:// URLs, a local TLS proxy is used to avoid GnuTLS issues.

    Note: this function intentionally connects to user-configured URLs, the same
    as the MJPEG and snapshot paths. The URL is sanitized and dangerous
    destinations are blocked before it reaches ffmpeg.
    """
    ffmpeg = get_ffmpeg_path()
    if not ffmpeg:
        logger.error("ffmpeg not found - required for RTSP capture")
        return None

    # ffmpeg's -i accepts every protocol it was built with, so an unchecked URL
    # here is a request to any host and scheme the caller names, not merely to a
    # camera. Restricting the scheme to RTSP is what keeps this a camera fetch.
    safe_url = _sanitize_camera_url(url, ("rtsp", "rtsps"))
    if not safe_url:
        logger.error("Invalid RTSP URL: %s...", redact_url_credentials(url)[:50])
        return None

    # If rtsps://, use TLS proxy
    proxy_server = None
    effective_url = safe_url
    if safe_url.lower().startswith("rtsps://"):
        try:
            from urllib.parse import urlparse

            from backend.app.services.camera import close_tls_proxy, create_tls_proxy

            parsed = urlparse(safe_url)
            target_port = parsed.port or 322
            proxy_port, proxy_server = await create_tls_proxy(parsed.hostname, target_port)
            userinfo = ""
            if parsed.username:
                userinfo = parsed.username
                if parsed.password:
                    userinfo += f":{parsed.password}"
                userinfo += "@"
            # Points at loopback deliberately, and is built after the check
            # above rather than re-checked: the destination that mattered was
            # the one the caller named, and it has already been vetted.
            effective_url = f"rtsp://{userinfo}127.0.0.1:{proxy_port}{parsed.path}"
            if parsed.query:
                effective_url += f"?{parsed.query}"
        except Exception as e:
            logger.warning("Failed to create TLS proxy for RTSP capture, falling back: %s", e)
            effective_url = safe_url

    cmd = [
        ffmpeg,
        "-rtsp_transport",
        "tcp",
        # Belt and braces on the scheme check above: a demuxer that follows a
        # reference out of the stream cannot leave these protocols either.
        "-protocol_whitelist",
        _RTSP_PROTOCOL_WHITELIST,
        "-i",
        effective_url,
        "-frames:v",
        "1",
        "-f",
        "image2pipe",
        "-vcodec",
        "mjpeg",
        "-q:v",
        "2",
        "-",
    ]

    try:
        logger.debug("Running ffmpeg RTSP capture...")
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
        logger.debug(
            "ffmpeg returned: code=%s, stdout=%s bytes, stderr=%s bytes",
            process.returncode,
            len(stdout),
            len(stderr),
        )

        if process.returncode != 0:
            # The summariser masks the camera password the input URL carries.
            logger.error("ffmpeg RTSP capture failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT)
            return None

        if not stdout or len(stdout) < 100:
            logger.error("ffmpeg returned empty or too small frame")
            return None

        return stdout

    except TimeoutError:
        logger.warning("RTSP frame capture timed out after %ss", timeout)
        if process:
            process.kill()
        return None
    except OSError as e:
        logger.error("RTSP frame capture failed: %s", e)
        return None
    finally:
        if proxy_server:
            await close_tls_proxy(proxy_server)


def _transcode_to_jpeg(data: bytes) -> bytes | None:
    """Decode an arbitrary still image (PNG/WebP/BMP/GIF/...) and re-encode as JPEG.

    Some camera/proxy snapshot endpoints serve stills as PNG or WebP rather than
    JPEG. A browser opened directly at the URL renders those fine, but our MJPEG
    ``multipart/x-mixed-replace`` stream hard-labels every part
    ``Content-Type: image/jpeg`` — so a non-JPEG payload makes the browser reject
    the frame and drop the whole stream ("connection lost", #1902). Transcoding to
    JPEG keeps the stream genuinely MJPEG and also keeps the JPEG-only downstream
    (plate detection, Obico, finish photo) working.

    Returns None if the bytes are not a decodable image (e.g. an HTML error page)
    or if the imaging libraries are unavailable — callers fall back to the raw
    bytes so behaviour is never worse than before.
    """
    try:
        import cv2
        import numpy as np
    except ImportError:
        return None
    try:
        img = cv2.imdecode(np.frombuffer(data, dtype=np.uint8), cv2.IMREAD_COLOR)
        if img is None:
            return None
        ok, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 85])
        if not ok:
            return None
        return buf.tobytes()
    except Exception as e:  # cv2 raises cv2.error (a subclass of Exception) on bad input
        logger.debug("Snapshot transcode to JPEG failed: %s", e)
        return None


async def _capture_snapshot(url: str, timeout: int) -> bytes | None:
    """Fetch snapshot from HTTP URL.

    Note: This function intentionally makes requests to user-configured URLs.
    External camera support requires connecting to user-specified camera endpoints.
    URL is sanitized and dangerous destinations are blocked.
    """
    # Sanitize URL - returns reconstructed URL from validated components
    safe_url = _sanitize_camera_url(url, ("http", "https"))
    if not safe_url:
        logger.error("Invalid snapshot URL format: %s...", redact_url_credentials(url)[:50])
        return None

    try:
        async with (
            aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session,
            session.get(safe_url) as response,
        ):
            if response.status != 200:
                logger.error("Snapshot URL returned status %s", response.status)
                return None

            data = await response.read()
    except TimeoutError:
        logger.warning("Snapshot capture timed out after %ss", timeout)
        return None
    except (aiohttp.ClientError, OSError) as e:
        logger.error("Snapshot capture failed: %s", e)
        return None

    # Fast path: already JPEG (SOI marker), stream it as-is (no decode/re-encode).
    if data.startswith(b"\xff\xd8"):
        return data

    # Not JPEG. Many snapshot endpoints serve PNG/WebP/BMP — transcode to JPEG so
    # the browser's MJPEG stream (and JPEG-only downstream) keep working instead of
    # dropping the connection (#1902). Run off the event loop: cv2 decode/encode is
    # CPU-bound and this can be polled at up to 15 fps while a camera view is open.
    transcoded = await asyncio.to_thread(_transcode_to_jpeg, data)
    if transcoded is not None:
        logger.debug(
            "Transcoded non-JPEG snapshot (%d bytes, header %s) to JPEG",
            len(data),
            data[:4].hex(),
        )
        return transcoded

    # Couldn't decode it as an image at all — most likely not an image response
    # (HTML error page, auth redirect, wrong URL). Return the raw bytes as a last
    # resort (unchanged behaviour) but log enough to debug.
    logger.warning(
        "External camera snapshot is not a decodable image "
        "(%d bytes, header %s) — verify the camera URL returns an image",
        len(data),
        data[:4].hex(),
    )
    return data


async def test_connection(url: str, camera_type: str) -> dict:
    """Test camera connection.

    Returns:
        Dict with {success: bool, error?: str, resolution?: str, coalesced: bool}

    ``coalesced`` is True when the frame came from a capture that was already
    running rather than from a connection this test opened. Captures are shared
    (see ``capture_frame``), so a test that lands while Obico is polling — or
    while any other one-shot consumer is mid-capture — gets that frame back and
    would otherwise report a healthy connection it never made, which is the one
    answer a *connection test* must not give silently. Forcing an uncoalesced
    capture here would be worse: it would open the second handle to a
    single-reader device that this whole mechanism exists to prevent. So the
    test still shares, and says so. Mirrors the ``coalesced_capture`` code the
    built-in diagnostic reports for the same situation (camera_diagnose.py).
    """
    logger.info("Testing camera connection: type=%s, url=%s...", camera_type, redact_url_credentials(url)[:50])
    # Sampled before the call, while it can still distinguish "someone else is
    # mid-capture" from "I am the one capturing".
    coalesced = capture_in_flight(url, camera_type)
    try:
        frame = await capture_frame(url, camera_type, timeout=10)
        logger.info("Capture result: %s bytes%s", len(frame) if frame else 0, " (coalesced)" if coalesced else "")

        if frame:
            # Try to get resolution from JPEG header
            resolution = None
            try:
                # Simple JPEG dimension extraction
                # SOF0 marker is FF C0, followed by length, precision, height, width
                sof_markers = [b"\xff\xc0", b"\xff\xc1", b"\xff\xc2"]
                for marker in sof_markers:
                    idx = frame.find(marker)
                    if idx != -1 and idx + 9 <= len(frame):
                        height = (frame[idx + 5] << 8) | frame[idx + 6]
                        width = (frame[idx + 7] << 8) | frame[idx + 8]
                        resolution = f"{width}x{height}"
                        break
            except (IndexError, ValueError):
                pass  # Resolution detection is optional; fall back to default

            return {"success": True, "resolution": resolution, "coalesced": coalesced}
        else:
            return {"success": False, "error": "Failed to capture frame from camera", "coalesced": coalesced}

    except Exception as e:
        # Sanitize error message - don't expose internal details
        error_type = type(e).__name__
        logger.error("Camera connection test failed: %s", e)
        return {"success": False, "error": f"Connection failed: {error_type}", "coalesced": coalesced}


async def generate_mjpeg_stream(
    url: str,
    camera_type: str,
    fps: int = 10,
    *,
    on_process: Callable[[asyncio.subprocess.Process], None] | None = None,
    on_frame: Callable[[bytes], None] | None = None,
    stop_event: asyncio.Event | None = None,
) -> AsyncGenerator[bytes, None]:
    """Generator yielding MJPEG frames for streaming.

    Args:
        url: Camera URL or USB device path
        camera_type: "mjpeg", "rtsp", "snapshot", or "usb"
        fps: Target frames per second
        on_process: Called with the spawned ffmpeg process for the ``usb`` and
            ``rtsp`` paths so the route layer can register it into the shared
            stream registries — that's what lets ``/camera/stop`` and the orphan
            janitor find and kill a leaked ffmpeg that's holding a USB device
            open (#2675). Without it the process is reachable only from this
            generator's own ``finally``, which an abrupt client disconnect can
            skip (same cancellation-timing class as #776).
        on_frame: Called with each RAW frame, before it is wrapped for the wire,
            so the route layer can publish it as the printer's buffered frame
            (#2707). It has to be a callback: what this generator yields is
            multipart-wrapped, so a consumer of the stream cannot recover the
            JPEG, and until now nothing populated the buffer for external
            cameras at all — leaving every one-shot consumer (layer timelapse,
            finish photo, Obico, plate check) with nothing to reuse and no
            option but to open a competing handle on a single-reader device.
            Exceptions are logged and swallowed: buffering must never be able
            to break the live stream.
        stop_event: When set, the reconnect loops stop retrying — so an explicit
            stop (which kills the current ffmpeg) doesn't immediately respawn a
            new process and reacquire the device.

    Yields:
        MJPEG frame data with HTTP multipart boundaries
    """
    frame_interval = 1.0 / max(fps, 1)
    last_frame_time = 0.0

    def _publish(frame: bytes) -> bytes:
        """Hand the raw frame to on_frame, then format it for the wire."""
        if on_frame is not None:
            try:
                on_frame(frame)
            except Exception:
                logger.exception("on_frame callback raised")
        return _format_mjpeg_frame(frame)

    if camera_type == "mjpeg":
        # Proxy MJPEG stream directly, with reconnect on timeout
        max_retries = 3
        for attempt in range(max_retries + 1):
            frame_yielded = False
            async for frame in _stream_mjpeg(url):
                frame_yielded = True
                current_time = asyncio.get_event_loop().time()
                if current_time - last_frame_time >= frame_interval:
                    last_frame_time = current_time
                    yield _publish(frame)
            if not frame_yielded or attempt == max_retries or (stop_event is not None and stop_event.is_set()):
                break
            logger.warning(
                "External MJPEG stream ended, reconnecting (attempt %d/%d)...",
                attempt + 1,
                max_retries,
            )
            await asyncio.sleep(2)

    elif camera_type == "rtsp":
        # Use ffmpeg to convert RTSP to MJPEG, with reconnect on timeout
        max_retries = 3
        for attempt in range(max_retries + 1):
            frame_yielded = False
            async for frame in _stream_rtsp(url, fps, on_process=on_process):
                frame_yielded = True
                yield _publish(frame)
            if not frame_yielded or attempt == max_retries or (stop_event is not None and stop_event.is_set()):
                break
            logger.warning(
                "External RTSP stream ended, reconnecting (attempt %d/%d)...",
                attempt + 1,
                max_retries,
            )
            await asyncio.sleep(2)

    elif camera_type == "usb":
        # Use ffmpeg to stream from USB camera
        async for frame in _stream_usb(url, fps, on_process=on_process):
            yield _publish(frame)

    elif camera_type == "snapshot":
        # Poll snapshot URL at interval
        while True:
            try:
                frame = await _capture_snapshot(url, timeout=10)
                if frame:
                    yield _publish(frame)
                await asyncio.sleep(frame_interval)
            except asyncio.CancelledError:
                break
            except (aiohttp.ClientError, OSError) as e:
                logger.warning("Snapshot poll failed: %s", e)
                await asyncio.sleep(frame_interval)


def _format_mjpeg_frame(frame: bytes) -> bytes:
    """Format frame for MJPEG HTTP response."""
    return (
        b"--frame\r\n"
        b"Content-Type: image/jpeg\r\n"
        b"Content-Length: " + str(len(frame)).encode() + b"\r\n"
        b"\r\n" + frame + b"\r\n"
    )


async def _stream_mjpeg(url: str) -> AsyncGenerator[bytes, None]:
    """Stream frames from MJPEG URL.

    Note: This function intentionally makes requests to user-configured URLs.
    External camera support requires connecting to user-specified camera endpoints.
    URL is sanitized and dangerous destinations are blocked.
    """
    # Sanitize URL - returns reconstructed URL from validated components
    safe_url = _sanitize_camera_url(url, ("http", "https"))
    if not safe_url:
        logger.error("Invalid MJPEG stream URL: %s...", redact_url_credentials(url)[:50])
        return

    try:
        timeout = aiohttp.ClientTimeout(total=None, sock_read=30)
        async with aiohttp.ClientSession(timeout=timeout) as session, session.get(safe_url) as response:
            if response.status != 200:
                logger.error("MJPEG stream returned status %s", response.status)
                return

            buffer = b""
            jpeg_start = b"\xff\xd8"
            jpeg_end = b"\xff\xd9"

            async for chunk in response.content.iter_chunked(8192):
                buffer += chunk

                # Extract complete frames from buffer
                while True:
                    start_idx = buffer.find(jpeg_start)
                    if start_idx == -1:
                        buffer = buffer[-2:] if len(buffer) > 2 else buffer
                        break

                    if start_idx > 0:
                        buffer = buffer[start_idx:]

                    end_idx = buffer.find(jpeg_end, 2)
                    if end_idx == -1:
                        break

                    frame = buffer[: end_idx + 2]
                    buffer = buffer[end_idx + 2 :]
                    yield frame

    except asyncio.CancelledError:
        logger.info("MJPEG stream cancelled")
    except (aiohttp.ClientError, OSError) as e:
        logger.error("MJPEG stream error: %s", e)


async def _stream_rtsp(
    url: str,
    fps: int,
    *,
    on_process: Callable[[asyncio.subprocess.Process], None] | None = None,
) -> AsyncGenerator[bytes, None]:
    """Stream frames from RTSP URL via ffmpeg.

    For rtsps:// URLs, a local TLS proxy (Python OpenSSL) is used instead
    of relying on ffmpeg's GnuTLS backend, which has compatibility issues
    with some printer firmwares.

    Note: this function intentionally connects to user-configured URLs. The URL
    is sanitized and dangerous destinations are blocked before it reaches
    ffmpeg — see ``_capture_rtsp_frame``, which guards the one-shot path the
    same way.
    """
    ffmpeg = get_ffmpeg_path()
    if not ffmpeg:
        logger.error("ffmpeg not found - required for RTSP streaming")
        return

    from backend.app.services.camera import rtsp_socket_timeout_flag

    safe_url = _sanitize_camera_url(url, ("rtsp", "rtsps"))
    if not safe_url:
        logger.error("Invalid RTSP stream URL: %s...", redact_url_credentials(url)[:50])
        return

    # If the URL uses rtsps://, set up a TLS proxy so ffmpeg uses plain rtsp://
    proxy_server = None
    effective_url = safe_url
    if safe_url.lower().startswith("rtsps://"):
        try:
            from urllib.parse import urlparse

            from backend.app.services.camera import close_tls_proxy, create_tls_proxy

            parsed = urlparse(safe_url)
            target_port = parsed.port or 322
            proxy_port, proxy_server = await create_tls_proxy(parsed.hostname, target_port)
            # Rewrite URL: rtsps://user:pass@host:port/path → rtsp://user:pass@127.0.0.1:proxy/path
            userinfo = ""
            if parsed.username:
                userinfo = parsed.username
                if parsed.password:
                    userinfo += f":{parsed.password}"
                userinfo += "@"
            # Loopback by design, and built after the check above rather than
            # re-checked — see the same rewrite in _capture_rtsp_frame.
            effective_url = f"rtsp://{userinfo}127.0.0.1:{proxy_port}{parsed.path}"
            if parsed.query:
                effective_url += f"?{parsed.query}"
        except Exception as e:
            logger.warning("Failed to create TLS proxy for RTSP, falling back to direct: %s", e)
            effective_url = safe_url

    cmd = [
        ffmpeg,
        "-rtsp_transport",
        "tcp",
        "-rtsp_flags",
        "prefer_tcp",
        "-protocol_whitelist",
        _RTSP_PROTOCOL_WHITELIST,
        # Socket I/O timeout name varies by ffmpeg version (#1504); see
        # `rtsp_socket_timeout_flag()` in services.camera.
        f"-{rtsp_socket_timeout_flag()}",
        "30000000",
        "-buffer_size",
        "1024000",
        "-max_delay",
        "500000",
        "-probesize",
        "32",
        "-analyzeduration",
        "0",
        "-fflags",
        "nobuffer",
        "-flags",
        "low_delay",
        "-i",
        effective_url,
        "-f",
        "mjpeg",
        "-q:v",
        "5",
        "-r",
        str(fps),
        "-an",
        "-",
    ]

    process = None
    try:
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        # Register immediately — before the startup probe below — so a process
        # that hangs on connect (rather than exiting) is still reachable by the
        # stop endpoint / orphan janitor (#2675).
        if on_process is not None:
            on_process(process)

        # Brief check for immediate startup failures
        await asyncio.sleep(0.1)
        if process.returncode is not None:
            stderr = await process.stderr.read()
            # The summariser masks the camera password the input URL carries.
            logger.error(
                "ffmpeg RTSP stream failed immediately: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT
            )
            return

        buffer = b""
        jpeg_start = b"\xff\xd8"
        jpeg_end = b"\xff\xd9"

        while True:
            try:
                chunk = await asyncio.wait_for(process.stdout.read(8192), timeout=30.0)

                if not chunk:
                    break

                buffer += chunk

                # Extract complete frames
                while True:
                    start_idx = buffer.find(jpeg_start)
                    if start_idx == -1:
                        buffer = buffer[-2:] if len(buffer) > 2 else buffer
                        break

                    if start_idx > 0:
                        buffer = buffer[start_idx:]

                    end_idx = buffer.find(jpeg_end, 2)
                    if end_idx == -1:
                        break

                    frame = buffer[: end_idx + 2]
                    buffer = buffer[end_idx + 2 :]
                    yield frame

            except TimeoutError:
                logger.warning("RTSP stream read timeout")
                break

    except asyncio.CancelledError:
        logger.info("RTSP stream cancelled")
    except OSError as e:
        logger.error("RTSP stream error: %s", e)
    finally:
        if process and process.returncode is None:
            process.terminate()
            try:
                await asyncio.wait_for(process.wait(), timeout=2.0)
            except TimeoutError:
                process.kill()
                await process.wait()
        if proxy_server:
            await close_tls_proxy(proxy_server)


async def _stream_usb(
    device: str,
    fps: int,
    *,
    on_process: Callable[[asyncio.subprocess.Process], None] | None = None,
) -> AsyncGenerator[bytes, None]:
    """Stream frames from USB camera via ffmpeg."""
    ffmpeg = get_ffmpeg_path()
    if not ffmpeg:
        logger.error("ffmpeg not found - required for USB camera streaming")
        return

    # Same validation as the one-shot path: a prefix check accepted
    # /dev/video/../../<anything that exists>, which -f v4l2 would then refuse
    # rather than the check refusing it.
    safe_device = _safe_usb_device_path(device)
    if not safe_device:
        return
    device = safe_device

    # ffmpeg command to stream from USB camera (v4l2)
    cmd = [
        ffmpeg,
        "-f",
        "v4l2",
        "-framerate",
        str(fps),
        "-i",
        device,
        "-f",
        "mjpeg",
        "-q:v",
        "5",
        "-r",
        str(fps),
        "-",
    ]

    process = None
    try:
        logger.info("Starting USB camera stream from %s at %s fps", device, fps)
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        # Register immediately — before the startup probe below — so a process
        # that hangs in open()/ioctl on a still-locked device (rather than
        # exiting with a "busy" error) is still reachable by the stop endpoint /
        # orphan janitor (#2675).
        if on_process is not None:
            on_process(process)

        # Give ffmpeg a moment to start and check for immediate failures
        await asyncio.sleep(0.5)
        if process.returncode is not None:
            stderr = await process.stderr.read()
            logger.error(
                "ffmpeg USB stream failed immediately: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT
            )
            return

        buffer = b""
        jpeg_start = b"\xff\xd8"
        jpeg_end = b"\xff\xd9"

        while True:
            try:
                chunk = await asyncio.wait_for(process.stdout.read(8192), timeout=30.0)

                if not chunk:
                    break

                buffer += chunk

                # Extract complete frames
                while True:
                    start_idx = buffer.find(jpeg_start)
                    if start_idx == -1:
                        buffer = buffer[-2:] if len(buffer) > 2 else buffer
                        break

                    if start_idx > 0:
                        buffer = buffer[start_idx:]

                    end_idx = buffer.find(jpeg_end, 2)
                    if end_idx == -1:
                        break

                    frame = buffer[: end_idx + 2]
                    buffer = buffer[end_idx + 2 :]
                    yield frame

            except TimeoutError:
                logger.warning("USB stream read timeout")
                break

    except asyncio.CancelledError:
        logger.info("USB stream cancelled")
    except OSError as e:
        logger.error("USB stream error: %s", e)
    finally:
        if process and process.returncode is None:
            process.terminate()
            try:
                await asyncio.wait_for(process.wait(), timeout=2.0)
            except TimeoutError:
                process.kill()
                await process.wait()