fix: validate command input to prevent execution of untrusted input

examples/time_in_zone/scripts/stream_from_file.py

@@ -84,7 +84,13 @@ def run_command_in_thread(command: list) -> Thread:
 
 
 def run_command(command: list) -> int:
-    process = subprocess.run(command)  # noqa: S603 # TODO: Validate command input to prevent execution of untrusted input
+    allowed_commands = ["docker", "ffmpeg"]
+    if not command or command[0] not in allowed_commands:
+        raise ValueError(
+            f"Invalid command '{command[0] if command else ''}'. "
+            f"Allowed: {allowed_commands}"
+        )
+    process = subprocess.run(command)  # noqa: S603
     return process.returncode
 
 

pyproject.toml

@@ -158,9 +158,6 @@ lint.per-file-ignores."notebooks/**" = [
   "PT018", # Assertion should be broken down into multiple parts
   "S101",  # Use of `assert` detected
 ]
-lint.per-file-ignores."src/**" = [
-  "S101", # TODO: Replace asserts with proper error handling
-]
 lint.per-file-ignores."tests/**" = [
   "S101", # Use of `assert` detected
 ]

src/supervision/annotators/core.py

@@ -89,7 +89,7 @@ class _BaseLabelAnnotator(BaseAnnotator):
         color_lookup: The method used to determine the color of the label.
         text_color: The color to use for the label text.
         text_padding: The padding around the label text, in pixels.
-        text_anchor: The position of the text relative to the detection
+        text_position: The position of the text relative to the detection
             bounding box.
         text_offset: A tuple of 2D coordinates `(x, y)` to
             offset the text position from the anchor point, in pixels.
@@ -444,6 +444,8 @@ def __init__(
                 Options are `INDEX`, `CLASS`, `TRACK`.
         """
         self.color: Color | ColorPalette = _normalize_color_input(color)
+        if not 0.0 <= opacity <= 1.0:
+            raise ValueError("Opacity must be between 0.0 and 1.0")
         self.opacity = opacity
         self.color_lookup: ColorLookup = color_lookup
 
@@ -624,6 +626,8 @@ def __init__(
         """
         self.color: Color | ColorPalette = _normalize_color_input(color)
         self.color_lookup: ColorLookup = color_lookup
+        if not 0.0 <= opacity <= 1.0:
+            raise ValueError("Opacity must be between 0.0 and 1.0")
         self.opacity = opacity
 
     @ensure_cv2_image_for_class_method
@@ -722,6 +726,8 @@ def __init__(
                 Options are `INDEX`, `CLASS`, `TRACK`.
         """
         self.color: Color | ColorPalette = _normalize_color_input(color)
+        if not 0.0 <= opacity <= 1.0:
+            raise ValueError("Opacity must be between 0.0 and 1.0")
         self.opacity = opacity
         self.color_lookup: ColorLookup = color_lookup
         self.kernel_size: int = kernel_size
@@ -781,7 +787,8 @@ def annotate(
             self.color_lookup if custom_color_lookup is None else custom_color_lookup,
             collect_union=True,
         )
-        assert fmask is not None  # collect_union=True always returns an array
+        if fmask is None:
+            raise ValueError("fmask cannot be None when collect_union=True")
 
         colored_mask = cv2.blur(colored_mask, (self.kernel_size, self.kernel_size))
         colored_mask[fmask] = [0, 0, 0]
@@ -1392,11 +1399,12 @@ def _draw_labels(
         detections: Detections,
         custom_color_lookup: npt.NDArray[np.int_] | None,
     ) -> None:
-        assert len(labels) == len(label_properties) == len(detections), (
-            f"Number of label properties ({len(label_properties)}), "
-            f"labels ({len(labels)}) and detections ({len(detections)}) "
-            "do not match."
-        )
+        if not (len(labels) == len(label_properties) == len(detections)):
+            raise ValueError(
+                f"Number of label properties ({len(label_properties)}), "
+                f"labels ({len(labels)}) and detections ({len(detections)}) "
+                "do not match."
+            )
 
         color_lookup = (
             custom_color_lookup
@@ -1616,7 +1624,8 @@ def annotate(
 
             ```
         """
-        assert isinstance(scene, Image.Image)
+        if not isinstance(scene, Image.Image):
+            raise TypeError("scene must be a PIL Image")
         _validate_labels(labels, detections)
 
         draw = ImageDraw.Draw(scene)
@@ -1706,11 +1715,12 @@ def _draw_labels(
         detections: Detections,
         custom_color_lookup: npt.NDArray[np.int_] | None,
     ) -> None:
-        assert len(labels) == len(label_properties) == len(detections), (
-            f"Number of label properties ({len(label_properties)}), "
-            f"labels ({len(labels)}) and detections ({len(detections)}) "
-            "do not match."
-        )
+        if not (len(labels) == len(label_properties) == len(detections)):
+            raise ValueError(
+                f"Number of label properties ({len(label_properties)}), "
+                f"labels ({len(labels)}) and detections ({len(detections)}) "
+                "do not match."
+            )
         color_lookup = (
             custom_color_lookup
             if custom_color_lookup is not None
@@ -2688,7 +2698,8 @@ def annotate(
             if custom_values is not None:
                 value = custom_values[detection_idx]
             else:
-                assert detections.confidence is not None  # MyPy type hint
+                if detections.confidence is None:
+                    raise ValueError("detections.confidence cannot be None")
                 value = detections.confidence[detection_idx]
 
             color = resolve_color(
@@ -3159,7 +3170,8 @@ def annotate(
 
     @staticmethod
     def _use_obb(detections_1: Detections, detections_2: Detections) -> bool:
-        assert not detections_1.is_empty() or not detections_2.is_empty()
+        if detections_1.is_empty() and detections_2.is_empty():
+            raise ValueError("At least one of the detections must not be empty")
         is_obb_1 = ORIENTED_BOX_COORDINATES in detections_1.data
         is_obb_2 = ORIENTED_BOX_COORDINATES in detections_2.data
         return (
@@ -3170,7 +3182,8 @@ def _use_obb(detections_1: Detections, detections_2: Detections) -> bool:
 
     @staticmethod
     def _use_mask(detections_1: Detections, detections_2: Detections) -> bool:
-        assert not detections_1.is_empty() or not detections_2.is_empty()
+        if detections_1.is_empty() and detections_2.is_empty():
+            raise ValueError("At least one of the detections must not be empty")
         is_mask_1 = detections_1.mask is not None
         is_mask_2 = detections_2.mask is not None
         return (
@@ -3217,7 +3230,8 @@ def _mask_from_mask(
         mask = np.zeros(scene.shape[:2], dtype=np.bool_)
         if detections.is_empty():
             return mask
-        assert detections.mask is not None
+        if detections.mask is None:
+            raise ValueError("detections.mask cannot be None")
 
         for detections_mask in detections.mask:
             mask |= detections_mask.astype(np.bool_)

src/supervision/detection/core.py

@@ -1883,15 +1883,17 @@ def from_vlm(
         vlm = _validate_vlm_parameters(vlm, result, kwargs)
 
         if vlm == VLM.PALIGEMMA:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             xyxy, class_id, class_name = from_paligemma(result, **kwargs)
             data: dict[str, npt.NDArray[np.generic] | list[Any]] = {
                 CLASS_NAME_DATA_FIELD: class_name,
             }
             return cls(xyxy=xyxy, class_id=class_id, data=data)
 
         if vlm == VLM.QWEN_2_5_VL:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             xyxy, class_id, class_name = from_qwen_2_5_vl(result, **kwargs)
             data = {CLASS_NAME_DATA_FIELD: class_name}
             confidence_arr: npt.NDArray[np.floating[Any]] = np.ones(
@@ -1902,7 +1904,8 @@ def from_vlm(
             )
 
         if vlm == VLM.QWEN_3_VL:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             xyxy, class_id, class_name = from_qwen_3_vl(result, **kwargs)
             data = {CLASS_NAME_DATA_FIELD: class_name}
             confidence_arr = np.ones(len(xyxy), dtype=float)
@@ -1911,13 +1914,15 @@ def from_vlm(
             )
 
         if vlm == VLM.DEEPSEEK_VL_2:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             xyxy, class_id, class_name = from_deepseek_vl_2(result, **kwargs)
             data = {CLASS_NAME_DATA_FIELD: class_name}
             return cls(xyxy=xyxy, class_id=class_id, data=data)
 
         if vlm == VLM.FLORENCE_2:
-            assert isinstance(result, dict)
+            if not isinstance(result, dict):
+                raise TypeError("result must be a dict")
             xyxy, labels, mask, xyxyxyxy = from_florence_2(result, **kwargs)
             if len(xyxy) == 0:
                 empty = cls.empty()
@@ -1933,18 +1938,21 @@ def from_vlm(
             return cls(xyxy=xyxy, mask=mask, data=data)
 
         if vlm == VLM.GOOGLE_GEMINI_2_0:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             xyxy, class_id, class_name = from_google_gemini_2_0(result, **kwargs)
             data = {CLASS_NAME_DATA_FIELD: class_name}
             return cls(xyxy=xyxy, class_id=class_id, data=data)
 
         if vlm == VLM.MOONDREAM:
-            assert isinstance(result, dict)
+            if not isinstance(result, dict):
+                raise TypeError("result must be a dict")
             xyxy = from_moondream(result, **kwargs)
             return cls(xyxy=xyxy)
 
         if vlm == VLM.GOOGLE_GEMINI_2_5:
-            assert isinstance(result, str)
+            if not isinstance(result, str):
+                raise TypeError("result must be a string")
             gemini_result = from_google_gemini_2_5(result, **kwargs)
             data = {CLASS_NAME_DATA_FIELD: gemini_result[2]}
             return cls(
@@ -2487,23 +2495,25 @@ def with_nms(
                 after non-maximum suppression.
 
         Raises:
-            AssertionError: If `confidence` is None and class_agnostic is False.
+            ValueError: If `confidence` is None and class_agnostic is False.
                 If `class_id` is None and class_agnostic is False.
         """
         if len(self) == 0:
             return self
 
-        assert self.confidence is not None, (
-            "Detections confidence must be given for NMS to be executed."
-        )
+        if self.confidence is None:
+            raise ValueError(
+                "Detections confidence must be given for NMS to be executed."
+            )
 
         if class_agnostic:
             predictions = np.hstack((self.xyxy, self.confidence.reshape(-1, 1)))
         else:
-            assert self.class_id is not None, (
-                "Detections class_id must be given for NMS to be executed. If you"
-                " intended to perform class agnostic NMS set class_agnostic=True."
-            )
+            if self.class_id is None:
+                raise ValueError(
+                    "Detections class_id must be given for NMS to be executed. If you"
+                    " intended to perform class agnostic NMS set class_agnostic=True."
+                )
             predictions = np.hstack(
                 (
                     self.xyxy,
@@ -2574,25 +2584,27 @@ def with_nmm(
             Groups of size 1 keep the original OBB unchanged.
 
         Raises:
-            AssertionError: If `confidence` is None or `class_id` is None and
+            ValueError: If `confidence` is None or `class_id` is None and
                 class_agnostic is False.
 
         ![non-max-merging](https://media.roboflow.com/supervision-docs/non-max-merging.png){ align=center width="800" }
         """  # noqa: E501 // docs
         if len(self) == 0:
             return self
 
-        assert self.confidence is not None, (
-            "Detections confidence must be given for NMM to be executed."
-        )
+        if self.confidence is None:
+            raise ValueError(
+                "Detections confidence must be given for NMM to be executed."
+            )
 
         if class_agnostic:
             predictions = np.hstack((self.xyxy, self.confidence.reshape(-1, 1)))
         else:
-            assert self.class_id is not None, (
-                "Detections class_id must be given for NMM to be executed. If you"
-                " intended to perform class agnostic NMM set class_agnostic=True."
-            )
+            if self.class_id is None:
+                raise ValueError(
+                    "Detections class_id must be given for NMM to be executed. If you"
+                    " intended to perform class agnostic NMM set class_agnostic=True."
+                )
             predictions = np.hstack(
                 (
                     self.xyxy,
@@ -2817,8 +2829,8 @@ def merge_inner_detection_object_pair(
     if detections_1.confidence is None and detections_2.confidence is None:
         merged_confidence = None
     else:
-        assert detections_1.confidence is not None
-        assert detections_2.confidence is not None
+        if detections_1.confidence is None or detections_2.confidence is None:
+            raise ValueError("confidence cannot be None")
         detection_1_area = (xyxy_1[2] - xyxy_1[0]) * (xyxy_1[3] - xyxy_1[1])
         detections_2_area = (xyxy_2[2] - xyxy_2[0]) * (xyxy_2[3] - xyxy_2[1])
         merged_confidence = (

src/supervision/detection/line_zone.py

@@ -281,8 +281,10 @@ def _compute_anchor_sides(
                 The third array, `has_any_right_trigger`, indicates if the
                 detection's anchor is on the right side of the line zone.
         """
-        assert len(detections) > 0
-        assert detections.tracker_id is not None
+        if len(detections) == 0:
+            raise ValueError("Detections must not be empty.")
+        if detections.tracker_id is None:
+            raise ValueError("Detections must have tracker_id.")
 
         all_anchors = np.array(
             [
@@ -312,7 +314,8 @@ def _update_class_id_to_name(self, detections: Detections) -> None:
         Assumes that class_names are only provided when class_ids are.
         """
         class_names = detections.data.get(CLASS_NAME_DATA_FIELD)
-        assert class_names is None or detections.class_id is not None
+        if class_names is not None and detections.class_id is None:
+            raise ValueError("class_names provided but detections.class_id is None.")
 
         if detections.class_id is None:
             return
@@ -618,7 +621,8 @@ def _draw_oriented_label(
             text_box_color=self.color,
             line_angle_degrees=line_angle_degrees,
         )
-        assert label_image.shape[0] == label_image.shape[1]
+        if label_image.shape[0] != label_image.shape[1]:
+            raise ValueError("Label image must be square.")
 
         text_width, text_height = cv2.getTextSize(
             text, cv2.FONT_HERSHEY_SIMPLEX, self.text_scale, self.text_thickness

src/supervision/detection/utils/converters.py

@@ -712,8 +712,10 @@ def mask_to_rle(
     ![mask_to_rle](https://media.roboflow.com/supervision-docs/
     mask-to-rle.png){ align=center width="800" }
     """
-    assert mask.ndim == 2, "Input mask must be 2D"
-    assert mask.size != 0, "Input mask cannot be empty"
+    if mask.ndim != 2:
+        raise ValueError("Input mask must be 2D")
+    if mask.size == 0:
+        raise ValueError("Input mask cannot be empty")
 
     counts: list[int] = cast(list[int], _mask_to_rle_counts(mask).tolist())
     if compressed: