diff --git a/inference/core/interfaces/camera/rtsp_opencv_tls.py b/inference/core/interfaces/camera/rtsp_opencv_tls.py index 952d6a2aa6..8c1173368a 100644 --- a/inference/core/interfaces/camera/rtsp_opencv_tls.py +++ b/inference/core/interfaces/camera/rtsp_opencv_tls.py @@ -1,11 +1,22 @@ -"""OpenCV/FFmpeg RTSPS TLS option builder (ENT-1544 B3).""" +"""OpenCV/FFmpeg RTSPS TLS option builder (ENT-1544 B3). + +Environment variables (also set by rfdm on Roboflow Edge devices): + +* ``ROBOFLOW_RTSP_TLS_VALIDATION_FLAGS`` — ``127`` strict (default when unset), + ``126`` or ``0`` to disable certificate verification (self-signed / custom CA). +* ``GST_SSL_CA_CERTIFICATE`` or ``SSL_CERT_FILE`` — PEM bundle for ``cafile``. + +Self-hosted / OSS deployments without rfdm must set these explicitly on the +inference container. Unmanaged ``rtsps://`` sources with self-signed certs need +``ROBOFLOW_RTSP_TLS_VALIDATION_FLAGS=126`` or the open will fail after this wiring. +""" from __future__ import annotations import os import threading from contextlib import contextmanager -from typing import Dict, Iterator, Optional +from typing import Dict, Iterator, Optional, Union from inference.core.interfaces.camera.rtsp_tls import ( GST_SSL_CA_CERTIFICATE_ENV_VAR, @@ -67,7 +78,7 @@ def merge_opencv_ffmpeg_capture_options( return _format_opencv_ffmpeg_capture_options(merged) -def build_opencv_ffmpeg_capture_options(video: str) -> Optional[str]: +def build_opencv_ffmpeg_capture_options(video: Union[str, int]) -> Optional[str]: """Build OPENCV_FFMPEG_CAPTURE_OPTIONS for RTSPS sources.""" if not is_rtsps_url(video): return None @@ -80,12 +91,18 @@ def build_opencv_ffmpeg_capture_options(video: str) -> Optional[str]: @contextmanager -def opencv_rtsps_tls_env(video: str) -> Iterator[None]: +def opencv_rtsps_tls_env(video: Union[str, int]) -> Iterator[None]: """Hold OPENCV_FFMPEG_CAPTURE_OPTIONS for one VideoCapture open. OpenCV reads this env var at capture-open time. The lock spans set → VideoCapture open → restore so concurrent RTSPS sources cannot clobber each other's TLS options. + + Non-RTSPS opens (USB, V4L2, files, plain ``rtsp://``) do not take this lock + and are not serialized here. A process-wide lock around every VideoCapture + would stall those backends, which have no FFmpeg open timeout. Concurrent + non-RTSPS FFmpeg opens may briefly inherit these options; that window is + preferred over blocking every producer on one hung camera. """ options = build_opencv_ffmpeg_capture_options(video) if options is None: diff --git a/inference/core/interfaces/camera/video_source.py b/inference/core/interfaces/camera/video_source.py index a22446b3dd..9e3dc64089 100644 --- a/inference/core/interfaces/camera/video_source.py +++ b/inference/core/interfaces/camera/video_source.py @@ -36,6 +36,7 @@ EndOfStreamError, StreamOperationNotAllowedError, ) +from inference.core.interfaces.camera.rtsp_opencv_tls import opencv_rtsps_tls_env from inference.core.interfaces.camera.source_reference_sanitizer import ( redact_credentials_in_text, sanitize_source_reference, @@ -149,7 +150,7 @@ class CV2VideoFrameProducer(VideoFrameProducer): def __init__(self, video: Union[str, int]): self._source_ref = video self._connection_error_message = "" - with capture_process_stderr() as captured_stderr: + with opencv_rtsps_tls_env(video), capture_process_stderr() as captured_stderr: if _consumes_camera_on_jetson(video=video): self.stream = cv2.VideoCapture(video, cv2.CAP_V4L2) else: diff --git a/tests/inference/unit_tests/core/interfaces/camera/test_video_source.py b/tests/inference/unit_tests/core/interfaces/camera/test_video_source.py index c557c6cc01..b5d5d4dc10 100644 --- a/tests/inference/unit_tests/core/interfaces/camera/test_video_source.py +++ b/tests/inference/unit_tests/core/interfaces/camera/test_video_source.py @@ -1,4 +1,5 @@ import importlib +import os import time from datetime import datetime from functools import partial @@ -25,6 +26,10 @@ SourceConnectionError, StreamOperationNotAllowedError, ) +from inference.core.interfaces.camera.rtsp_opencv_tls import ( + OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, +) +from inference.core.interfaces.camera.rtsp_tls import RTSP_TLS_VALIDATION_FLAGS_ENV_VAR from inference.core.interfaces.camera.stream_error_codes import StreamErrorCode from inference.core.interfaces.camera.video_source import ( BufferConsumptionStrategy, @@ -1917,3 +1922,114 @@ def test_consume_video_emits_poison_pill_on_consume_error() -> None: assert source._state is StreamState.ERROR with pytest.raises(EndOfStreamError): source.read_frame(timeout=0.0) + + +def test_cv2_producer_sets_ffmpeg_tls_options_only_for_rtsps( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # given + monkeypatch.delenv(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, raising=False) + monkeypatch.delenv(RTSP_TLS_VALIDATION_FLAGS_ENV_VAR, raising=False) + seen = {} + + def fake_video_capture(video, *args, **kwargs): + seen[video] = os.environ.get(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR) + return MagicMock() + + # when + with patch.object(video_source.cv2, "VideoCapture", fake_video_capture): + CV2VideoFrameProducer("rtsps://camera.example/stream") + CV2VideoFrameProducer("rtsp://camera.example/stream") + + # then - OpenCV reads the env var at capture-open time, so it must be set + # while VideoCapture runs for RTSPS, and left alone for plain RTSP + assert "tls_verify;1" in seen["rtsps://camera.example/stream"] + assert seen["rtsp://camera.example/stream"] is None + assert OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR not in os.environ + + +def test_cv2_producer_restores_ffmpeg_tls_env_after_open( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, raising=False) + monkeypatch.delenv(RTSP_TLS_VALIDATION_FLAGS_ENV_VAR, raising=False) + + with patch.object(video_source.cv2, "VideoCapture", MagicMock()): + CV2VideoFrameProducer("rtsps://camera.example/stream") + + assert OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR not in os.environ + + +def test_cv2_producer_restores_ffmpeg_tls_env_when_capture_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, raising=False) + monkeypatch.delenv(RTSP_TLS_VALIDATION_FLAGS_ENV_VAR, raising=False) + + def failing_video_capture(*args, **kwargs): + raise RuntimeError("simulated open failure") + + with patch.object( + video_source.cv2, "VideoCapture", failing_video_capture + ), pytest.raises(RuntimeError): + CV2VideoFrameProducer("rtsps://camera.example/stream") + + assert OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR not in os.environ + + +def test_cv2_producer_leaves_ffmpeg_tls_env_unset_for_device_index( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, raising=False) + seen = {} + + def fake_video_capture(video, *args, **kwargs): + seen["video"] = video + seen["env"] = os.environ.get(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR) + return MagicMock() + + with patch.object(video_source.cv2, "VideoCapture", fake_video_capture): + CV2VideoFrameProducer(0) + + assert seen["video"] == 0 + assert seen["env"] is None + assert OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR not in os.environ + + +def test_cv2_producer_non_rtsps_open_does_not_wait_on_stuck_rtsps( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(OPENCV_FFMPEG_CAPTURE_OPTIONS_ENV_VAR, raising=False) + monkeypatch.delenv(RTSP_TLS_VALIDATION_FLAGS_ENV_VAR, raising=False) + rtsps_entered = Event() + release_rtsps = Event() + non_rtsps_opened = Event() + + def fake_video_capture(video, *args, **kwargs): + if isinstance(video, str) and video.startswith("rtsps://"): + rtsps_entered.set() + assert release_rtsps.wait(timeout=2.0) + else: + non_rtsps_opened.set() + return MagicMock() + + with patch.object(video_source.cv2, "VideoCapture", fake_video_capture): + rtsps_thread = Thread( + target=lambda: CV2VideoFrameProducer("rtsps://camera.example/stream") + ) + rtsps_thread.start() + assert rtsps_entered.wait(timeout=1.0) + + non_rtsps_thread = Thread(target=lambda: CV2VideoFrameProducer(0)) + non_rtsps_thread.start() + opened_without_waiting = non_rtsps_opened.wait(timeout=1.0) + + release_rtsps.set() + rtsps_thread.join(timeout=2.0) + non_rtsps_thread.join(timeout=2.0) + + assert opened_without_waiting, ( + "USB/file/V4L2 opens must not wait on a stuck RTSPS VideoCapture" + ) + assert not rtsps_thread.is_alive() + assert not non_rtsps_thread.is_alive()