ENT-1677: Apply RTSPS TLS options to the OpenCV capture open

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:

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:

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()