From f0c5b80e1e3d15a539fbda9688366342857a6a1d Mon Sep 17 00:00:00 2001 From: NVergunst-ROBO Date: Mon, 17 Aug 2026 22:30:43 -0400 Subject: [PATCH 1/4] ENT-1677: Apply RTSPS TLS options to the OpenCV capture open Wire opencv_rtsps_tls_env into CV2VideoFrameProducer so non-Jetson OpenCV/FFmpeg RTSPS opens honor rfdm-injected CA bundle and TLS flags. - Enter TLS env before capture_process_stderr (lock outside stderr redirect). - Widen helper signatures to Union[str, int] for the producer call site. - Document RTSPS-vs-RTSPS env isolation in the helper docstring. --- .../core/interfaces/camera/rtsp_opencv_tls.py | 9 +++--- .../core/interfaces/camera/video_source.py | 3 +- .../interfaces/camera/test_video_source.py | 28 +++++++++++++++++++ 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/inference/core/interfaces/camera/rtsp_opencv_tls.py b/inference/core/interfaces/camera/rtsp_opencv_tls.py index 952d6a2aa6..e6d35b3016 100644 --- a/inference/core/interfaces/camera/rtsp_opencv_tls.py +++ b/inference/core/interfaces/camera/rtsp_opencv_tls.py @@ -5,7 +5,7 @@ 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 +67,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 +80,13 @@ 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. + other's TLS options. Isolation is RTSPS-vs-RTSPS only; concurrent non-RTSPS + opens do not take the lock and may briefly inherit these options. """ 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..6e88edc912 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,26 @@ 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 From b1fe82b48ab31285b1f124638587c081bba3166e Mon Sep 17 00:00:00 2001 From: NVergunst-ROBO Date: Thu, 20 Aug 2026 19:37:37 -0400 Subject: [PATCH 2/4] ENT-1677: Document RTSPS TLS env vars and extend producer tests Add module docstring for ROBOFLOW_RTSP_TLS_VALIDATION_FLAGS and CA bundle env vars (self-hosted escape hatch). Assert env var is restored after open, on capture failure, and untouched for int device indices. --- .../core/interfaces/camera/rtsp_opencv_tls.py | 13 ++++- .../interfaces/camera/test_video_source.py | 49 +++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/inference/core/interfaces/camera/rtsp_opencv_tls.py b/inference/core/interfaces/camera/rtsp_opencv_tls.py index e6d35b3016..92a6b79d1b 100644 --- a/inference/core/interfaces/camera/rtsp_opencv_tls.py +++ b/inference/core/interfaces/camera/rtsp_opencv_tls.py @@ -1,4 +1,15 @@ -"""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 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 6e88edc912..abb3e5604a 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 @@ -1945,3 +1945,52 @@ def fake_video_capture(video, *args, **kwargs): # 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 From 452dd80dc9f90ead6370af0874f3ea72a1c298cf Mon Sep 17 00:00:00 2001 From: NVergunst-ROBO Date: Fri, 21 Aug 2026 10:17:20 -0400 Subject: [PATCH 3/4] ENT-1677: Serialize VideoCapture open behind a process lock OPENCV_FFMPEG_CAPTURE_OPTIONS and stderr dup2 are process-global, so concurrent camera opens can clobber each other. Hold one lock around the open. --- inference/core/interfaces/camera/video_source.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/inference/core/interfaces/camera/video_source.py b/inference/core/interfaces/camera/video_source.py index 9e3dc64089..b6934069e1 100644 --- a/inference/core/interfaces/camera/video_source.py +++ b/inference/core/interfaces/camera/video_source.py @@ -146,11 +146,20 @@ def locked_executor(video_source: "VideoSource", *args, **kwargs) -> None: return locked_executor +# VideoCapture open mutates process-global state (OPENCV_FFMPEG_CAPTURE_OPTIONS +# and stderr fd 2). Serialize so concurrent cameras cannot clobber each other. +_capture_open_lock = Lock() + + class CV2VideoFrameProducer(VideoFrameProducer): def __init__(self, video: Union[str, int]): self._source_ref = video self._connection_error_message = "" - with opencv_rtsps_tls_env(video), capture_process_stderr() as captured_stderr: + with ( + _capture_open_lock, + 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: From cb2e61c4a110f76fb7571118bba6707ffc964a72 Mon Sep 17 00:00:00 2001 From: NVergunst-ROBO Date: Fri, 21 Aug 2026 16:15:52 -0400 Subject: [PATCH 4/4] ENT-1677: Do not serialize non-RTSPS VideoCapture opens A process-wide lock around every producer would stall USB/V4L2/file opens that have no FFmpeg timeout. Keep the RTSPS-only TLS env lock. --- .../core/interfaces/camera/rtsp_opencv_tls.py | 9 ++++- .../core/interfaces/camera/video_source.py | 11 +----- .../interfaces/camera/test_video_source.py | 39 +++++++++++++++++++ 3 files changed, 47 insertions(+), 12 deletions(-) diff --git a/inference/core/interfaces/camera/rtsp_opencv_tls.py b/inference/core/interfaces/camera/rtsp_opencv_tls.py index 92a6b79d1b..8c1173368a 100644 --- a/inference/core/interfaces/camera/rtsp_opencv_tls.py +++ b/inference/core/interfaces/camera/rtsp_opencv_tls.py @@ -96,8 +96,13 @@ def opencv_rtsps_tls_env(video: Union[str, int]) -> Iterator[None]: 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. Isolation is RTSPS-vs-RTSPS only; concurrent non-RTSPS - opens do not take the lock and may briefly inherit these options. + 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 b6934069e1..9e3dc64089 100644 --- a/inference/core/interfaces/camera/video_source.py +++ b/inference/core/interfaces/camera/video_source.py @@ -146,20 +146,11 @@ def locked_executor(video_source: "VideoSource", *args, **kwargs) -> None: return locked_executor -# VideoCapture open mutates process-global state (OPENCV_FFMPEG_CAPTURE_OPTIONS -# and stderr fd 2). Serialize so concurrent cameras cannot clobber each other. -_capture_open_lock = Lock() - - class CV2VideoFrameProducer(VideoFrameProducer): def __init__(self, video: Union[str, int]): self._source_ref = video self._connection_error_message = "" - with ( - _capture_open_lock, - opencv_rtsps_tls_env(video), - 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 abb3e5604a..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 @@ -1994,3 +1994,42 @@ def fake_video_capture(video, *args, **kwargs): 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()