chore: suppress Bandit false positives (#1561)

AlexanderBarabanov · web-flow · commit 0ae677817726 · 2025-11-27T15:43:34.000Z
Signed-off-by: Barabanov, Alexander &lt;alexander.barabanov@intel.com&gt;
diff --git a/dev_tools/pre_commit_scripts/find_py_projects_and_run_make_target.py b/dev_tools/pre_commit_scripts/find_py_projects_and_run_make_target.py
@@ -41,7 +41,7 @@
 exit_code = 0
 
 for project_dir, project_files in files_per_project.items():
-    if subprocess.run(
+    if subprocess.run( # nosec: B603
         ["make", "-n", make_target],
         cwd=project_dir,
         stdout=subprocess.DEVNULL,
@@ -55,7 +55,7 @@
         project_files = [
             str(Path(file).relative_to(project_dir)) for file in project_files
         ]
-        exit_code |= subprocess.run(
+        exit_code |= subprocess.run( # nosec: B603
             ["make", make_target, "FILES=" + " ".join(project_files)], cwd=project_dir
         ).returncode
 
diff --git a/interactive_ai/data_migration/migration/scripts/convert_vfr_videos_to_cfr.py b/interactive_ai/data_migration/migration/scripts/convert_vfr_videos_to_cfr.py
@@ -92,7 +92,7 @@ def _get_video_fps(cls, video_path: str) -> float:
         :param video_path: Local path or presigned S3 URL pointing to the video
         :return: Video FPS as float
         """
-        result = subprocess.run(  # noqa: S603
+        result = subprocess.run(  # noqa: S603 # nosec: B603
             [  # noqa: S607
                 "ffprobe",
                 "-v",
@@ -122,7 +122,7 @@ def _is_variable_frame_rate(cls, video_path: str) -> bool:
         :return: True if the video has variable frame rate, False if constant frame rate
         """
         # Get r_frame_rate
-        r_frame_rate_result = subprocess.run(  # noqa: S603
+        r_frame_rate_result = subprocess.run(  # noqa: S603 # nosec: B603
             [  # noqa: S607
                 "ffprobe",
                 "-v",
@@ -141,7 +141,7 @@ def _is_variable_frame_rate(cls, video_path: str) -> bool:
         )
 
         # Get avg_frame_rate
-        avg_frame_rate_result = subprocess.run(  # noqa: S603
+        avg_frame_rate_result = subprocess.run(  # noqa: S603 # nosec: B603
             [  # noqa: S607
                 "ffprobe",
                 "-v",
@@ -307,7 +307,7 @@ def _stitch_frames_to_cfr_video(cls, frames_dir: str, output_path: str, fps: flo
             logger.info(f"Stitching {len(frame_files)} frames into CFR video at {fps} fps")
 
             # Use ffmpeg to create video from image sequence
-            process = subprocess.run(  # noqa: S603
+            process = subprocess.run(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffmpeg",
                     "-y",  # Overwrite output file
diff --git a/interactive_ai/data_migration/migration/scripts/correct_vfr_video_fps.py b/interactive_ai/data_migration/migration/scripts/correct_vfr_video_fps.py
@@ -53,7 +53,7 @@ def upgrade_project(cls, organization_id: str, workspace_id: str, project_id: st
     @staticmethod
     def get_video_fps(video_path: str) -> float:
         # Run ffprobe to get video stream information
-        result = subprocess.run(  # noqa: S603
+        result = subprocess.run(  # noqa: S603 # nosec: B603
             [  # noqa: S607
                 "ffprobe",
                 "-v",
diff --git a/interactive_ai/libs/media_utils/media_utils/video_decoder.py b/interactive_ai/libs/media_utils/media_utils/video_decoder.py
@@ -172,7 +172,7 @@ def reset_reader(self, file_location: str) -> None:
 
     def get_fps(self, file_location: str) -> float:
         # Run ffprobe to get video stream information
-        result = subprocess.run(  # noqa: S603
+        result = subprocess.run(  # noqa: S603 # nosec: B603
             [  # noqa: S607
                 "ffprobe",
                 "-v",
@@ -207,7 +207,7 @@ def is_variable_frame_rate(self, file_location: str) -> bool:
         """
         try:
             # Get r_frame_rate
-            r_frame_rate_result = subprocess.run(  # noqa: S603
+            r_frame_rate_result = subprocess.run(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffprobe",
                     "-v",
@@ -226,7 +226,7 @@ def is_variable_frame_rate(self, file_location: str) -> bool:
             )
 
             # Get avg_frame_rate
-            avg_frame_rate_result = subprocess.run(  # noqa: S603
+            avg_frame_rate_result = subprocess.run(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffprobe",
                     "-v",
@@ -278,7 +278,7 @@ def convert_vfr_to_cfr(self, input_path: str, output_path: str, target_fps: floa
         :return: True if conversion was successful, False otherwise
         """
         try:
-            process = subprocess.run(  # noqa: S603
+            process = subprocess.run(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffmpeg",
                     "-i",
diff --git a/interactive_ai/libs/media_utils/media_utils/video_file_repair.py b/interactive_ai/libs/media_utils/media_utils/video_file_repair.py
@@ -65,7 +65,7 @@ def attempt_repair(video_binary_repo: VideoBinaryRepo, filename: str) -> bool:
 
             import subprocess
 
-            process = subprocess.Popen(  # noqa: S603
+            process = subprocess.Popen(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffmpeg",
                     "-hide_banner",
diff --git a/interactive_ai/libs/media_utils/media_utils/video_thumbnail.py b/interactive_ai/libs/media_utils/media_utils/video_thumbnail.py
@@ -32,7 +32,7 @@ def generate_thumbnail_video(
         target_resolution = f"{default_thumbnail_size}:-2"
 
     _, tmp_thumbnail_path = tempfile.mkstemp()
-    process = subprocess.Popen(  # noqa: S603
+    process = subprocess.Popen(  # noqa: S603 # nosec: B603
         [  # noqa: S607
             "ffmpeg",
             "-threads",
diff --git a/interactive_ai/services/model_registration/app/service/model_converter.py b/interactive_ai/services/model_registration/app/service/model_converter.py
@@ -12,7 +12,7 @@
 import zipfile
 from enum import Enum, auto
 from pathlib import Path
-from xml.etree.ElementTree import Element
+from xml.etree.ElementTree import Element # nosec: B405 # defusedxml is used to parse data
 
 from defusedxml import ElementTree
 from grpc_interfaces.model_registration.pb.service_pb2 import Model, Project
diff --git a/interactive_ai/services/resource/app/resource_management/media_manager.py b/interactive_ai/services/resource/app/resource_management/media_manager.py
@@ -731,7 +731,7 @@ def create_and_save_thumbnail_video(
                 filename=video.thumbnail_video_filename, make_unique=False
             )
             video_path_or_url = str(video_binary_repo.get_path_or_presigned_url(filename=video.data_binary_filename))
-            process = subprocess.Popen(  # noqa: S603
+            process = subprocess.Popen(  # noqa: S603 # nosec: B603
                 [  # noqa: S607
                     "ffmpeg",
                     "-threads",
diff --git a/interactive_ai/workflows/common/jobs_common_extras/datumaro_conversion/convert_utils.py b/interactive_ai/workflows/common/jobs_common_extras/datumaro_conversion/convert_utils.py
@@ -765,7 +765,7 @@ def build_labels_data(label_names: list[str], anomaly_label_names: list[str] | N
         for label_name in label_names:
             label_data: dict[str, Any] = {
                 "name": label_name,
-                "color": "#00" + "".join([random.choice("0123456789ABCDEF") for j in range(6)]),  # noqa: S311
+                "color": "#00" + "".join([random.choice("0123456789ABCDEF") for j in range(6)]),  # noqa: S311 # nosec: B311
             }
             label_datas.append(label_data)
 
diff --git a/interactive_ai/workflows/train/trainer/scripts/s3_client.py b/interactive_ai/workflows/train/trainer/scripts/s3_client.py
@@ -53,7 +53,7 @@ def wrapper(*args, **kwargs):
                         raise
 
                     # ruff: noqa: S311
-                    jitter = random.uniform(0, 1)  # nosec
+                    jitter = random.uniform(0, 1)  # nosec: B311
                     backoff_time = min(jitter * (2**retries) * delay, max_backoff)
                     time.sleep(backoff_time)
                     retries += 1
diff --git a/platform/services/installer/app/checks/os.py b/platform/services/installer/app/checks/os.py
@@ -51,7 +51,7 @@ def _get_local_os() -> str:
     try:
         logger.debug("Checking OS version with `grep ^PRETTY /etc/os-release`")
         return (
-            subprocess.check_output(["grep", "^PRETTY", "/etc/os-release"], timeout=5)  # noqa: S607
+            subprocess.check_output(["grep", "^PRETTY", "/etc/os-release"], timeout=5)  # noqa: S607 # nosec: B603
             .decode("utf-8")
             .strip()
             .replace("PRETTY_NAME=", "")
diff --git a/platform/services/installer/app/checks/resources.py b/platform/services/installer/app/checks/resources.py
@@ -69,7 +69,7 @@ def _bytes_to_gigabytes(bytes_: int) -> float:
 
 def _subprocess_run(command: list[str]):
     with open(INSTALL_LOG_FILE_PATH, mode="a+") as log_file:
-        process = subprocess.run(  # noqa: S603
+        process = subprocess.run(  # noqa: S603 # nosec: B603
             command,
             check=True,
             stdout=subprocess.PIPE,
@@ -100,7 +100,7 @@ def _check_intel_driver() -> bool:
     """
     try:
         logger.debug("Checking the installed display devices with `hwinfo --display`")
-        hwinfo_output = subprocess.check_output(["hwinfo", "--display"], timeout=5).decode("utf-8")  # noqa: S607
+        hwinfo_output = subprocess.check_output(["hwinfo", "--display"], timeout=5).decode("utf-8")  # noqa: S607 # nosec: B603
         logger.debug(hwinfo_output)
         if ResourcesChecksTexts.intel_gpu_driver_displayed in hwinfo_output:
             logger.debug('hwinfo returned at least one device with Driver: "i915" displayed')
@@ -143,7 +143,7 @@ def check_gpu_driver_version(config: InstallationConfig | UpgradeConfig) -> None
     if config.gpu_provider.value == GPU_PROVIDER_NVIDIA:
         try:
             nvidia_smi_output = subprocess.check_output(
-                ["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader,nounits"]  # noqa: S607
+                ["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader,nounits"]  # noqa: S607 # nosec: B603
             )
             nvidia_driver_versions = nvidia_smi_output.decode("utf-8").strip().split("\n")
             for nvidia_driver_version in nvidia_driver_versions:
@@ -169,7 +169,7 @@ def _check_intel_gpu_driver(env: dict[str, str]) -> bool:
         command = 'clinfo|grep "' + ResourcesChecksTexts.intel_gpu_arc_device_name + '"|grep Intel'
         logger.debug(f"Getting the list of Intel GPU drivers with {command}")
 
-        clinfo_output = subprocess.check_output(  # noqa: S602  # nosec: B602
+        clinfo_output = subprocess.check_output(  # noqa: S602 # nosec: B602
             command,
             stderr=subprocess.STDOUT,
             shell=True,
@@ -205,7 +205,7 @@ def _get_intel_gpus() -> tuple[str, bool]:  # noqa: C901
     # Only valid for MAX cards
     try:
         logger.debug("Getting the list of Intel GPU with `xpu-smi discovery`")
-        xpu_output = subprocess.check_output(["xpu-smi", "discovery"], timeout=5, env=env).decode("utf-8")  # noqa: S607
+        xpu_output = subprocess.check_output(["xpu-smi", "discovery"], timeout=5, env=env).decode("utf-8")  # noqa: S607 # nosec: B603
         logger.debug(xpu_output)
         if ResourcesChecksTexts.intel_gpu_no_devices in xpu_output:
             logger.debug("No devices")
@@ -224,7 +224,7 @@ def _get_intel_gpus() -> tuple[str, bool]:  # noqa: C901
         command = "lspci -nnk | grep -iA3 'VGA\|3D\|Display'"
         logger.debug(f"Getting the list of Intel ARC with {command}")
 
-        lspci_output = subprocess.check_output(  # noqa: S602  # nosec: B602
+        lspci_output = subprocess.check_output(  # noqa: S602 # nosec: B602
             command,
             stderr=subprocess.STDOUT,
             shell=True,
diff --git a/platform/services/installer/app/checks/tools.py b/platform/services/installer/app/checks/tools.py
@@ -20,7 +20,7 @@ def _is_snap_installed():
     """
     logger.debug("Checking if snap is installed")
     try:
-        subprocess.check_output(["snap", "--version"])  # noqa: S607
+        subprocess.check_output(["snap", "--version"])  # noqa: S607 # nosec: B603
         logger.debug("snap is installed.")
         return True
     except (subprocess.CalledProcessError, FileNotFoundError):
@@ -46,7 +46,7 @@ def _is_curl_installed():
     """
     logger.debug("Checking curl is installed")
     try:
-        subprocess.check_output(["curl", "--version"])  # noqa: S607
+        subprocess.check_output(["curl", "--version"])  # noqa: S607 # nosec: B603
         logger.debug("curl is installed.")
         return True
     except (subprocess.CalledProcessError, FileNotFoundError):
@@ -62,7 +62,7 @@ def _check_if_curl_has_internet_access():
         logger.info("Checking if curl has internet access.")
         logger.debug("Sending request to http://example.com...")
         subprocess.check_output(
-            ["curl", "--head", "--silent", "http://example.com", "--connect-timeout", "5"]  # noqa: S607
+            ["curl", "--head", "--silent", "http://example.com", "--connect-timeout", "5"]  # noqa: S607 # nosec: B603
         )
         logger.debug("curl has access to Internet.")
     except (subprocess.CalledProcessError, FileNotFoundError) as error:
diff --git a/platform/services/installer/app/cli_utils/platform_logs.py b/platform/services/installer/app/cli_utils/platform_logs.py
@@ -88,7 +88,7 @@ def subprocess_run(command, log_file, env=None, cwd=None):  # noqa: ANN001, ANN2
     """
     Runs the command and outputs to platform logs and/or stdout.
     """
-    with subprocess.Popen(  # noqa: S603
+    with subprocess.Popen(  # noqa: S603 # nosec: B603
         command,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
diff --git a/platform/services/installer/app/k3s/install.py b/platform/services/installer/app/k3s/install.py
@@ -100,7 +100,7 @@ def _k3s_ready(time_wait: int = 300, delay: int = 10) -> bool:
 
     for _ in range(0, time_wait, delay):
         try:
-            result = subprocess.run(  # noqa: S603
+            result = subprocess.run(  # noqa: S603 # nosec: B603
                 node_cmd,
                 capture_output=True,
                 text=True,
diff --git a/platform/services/installer/app/platform_stages/steps/apply_minor_version_changes/helm_keep_resource_policy.py b/platform/services/installer/app/platform_stages/steps/apply_minor_version_changes/helm_keep_resource_policy.py
@@ -42,7 +42,7 @@ def resource_exists(resource: ResourceToHandle) -> bool:
     if resource.namespace:
         check_cmd.extend(["-n", resource.namespace])
 
-    result = subprocess.run(check_cmd, capture_output=True, text=True, check=False)  # noqa: S603
+    result = subprocess.run(check_cmd, capture_output=True, text=True, check=False)  # noqa: S603 # nosec: B603
     return result.returncode == 0
 
 
diff --git a/platform/services/installer/app/platform_stages/steps/load_images.py b/platform/services/installer/app/platform_stages/steps/load_images.py
@@ -53,7 +53,7 @@ def _extract_version(pattern: str, image: str) -> str:
 
 def _run_subprocess(command: list[str], error_message: str) -> subprocess.CompletedProcess:
     try:
-        return subprocess.run(command, capture_output=True, text=True, check=True)  # noqa: S603
+        return subprocess.run(command, capture_output=True, text=True, check=True)  # noqa: S603 # nosec: B603
     except subprocess.CalledProcessError as e:
         logger.error(f"{error_message}: {e.stderr}")
         raise PinImageVersionError from e
diff --git a/platform/services/installer/tools/change_password.py b/platform/services/installer/tools/change_password.py
@@ -25,7 +25,7 @@ def _get_pod_name(kubeconfig: str) -> str:
     ]
     try:
         logger.debug(f"Running command: {' '.join(pod_name_cmd)}")
-        pod_name = subprocess.run(pod_name_cmd, capture_output=True, text=True, check=True)  # noqa: S603
+        pod_name = subprocess.run(pod_name_cmd, capture_output=True, text=True, check=True)  # noqa: S603 # nosec: B603
         return pod_name.stdout.strip()
     except subprocess.CalledProcessError as e:
         logger.exception(f"Error: {e.stderr}")
@@ -51,7 +51,7 @@ def _change_password(username: str, password: str, kubeconfig: str, pod_name: st
     ]
     try:
         logger.debug(f"Running command: {' '.join(kubectl_cmd)}")
-        subprocess.run(kubectl_cmd, text=True, check=True, stderr=subprocess.STDOUT)  # noqa: S603
+        subprocess.run(kubectl_cmd, text=True, check=True, stderr=subprocess.STDOUT)  # noqa: S603 # nosec: B603
 
     except subprocess.CalledProcessError as e:
         logger.exception(f"Error: {e.stderr}")
diff --git a/platform/services/weights_uploader/app/pretrained_models/pretrained_models.py b/platform/services/weights_uploader/app/pretrained_models/pretrained_models.py
@@ -38,7 +38,7 @@ def download_file(url: str, target_path: str, auto_unzip=True):  # noqa: ANN001,
     download_temp_target_path = os.path.join(target_dir_path, url_original_filename)
 
     with (
-        urllib.request.urlopen(url) as response,  # noqa: S310
+        urllib.request.urlopen(url) as response,  # noqa: S310 # nosec: B310 # URLs hardcoded in JSON
         open(download_temp_target_path, "wb") as out_file,
     ):
         shutil.copyfileobj(response, out_file)

Original file line number	Diff line number	Diff line change
`@@ -731,7 +731,7 @@ def create_and_save_thumbnail_video(`
`731`	`731`	`filename=video.thumbnail_video_filename, make_unique=False`
`732`	`732`	`)`
`733`	`733`	`video_path_or_url = str(video_binary_repo.get_path_or_presigned_url(filename=video.data_binary_filename))`
`734`		`- process = subprocess.Popen( # noqa: S603`
	`734`	`+ process = subprocess.Popen( # noqa: S603 # nosec: B603`
`735`	`735`	`[ # noqa: S607`
`736`	`736`	`"ffmpeg",`
`737`	`737`	`"-threads",`
Original file line number	Diff line number	Diff line change
`@@ -765,7 +765,7 @@ def build_labels_data(label_names: list[str], anomaly_label_names: list[str] \| N`
`765`	`765`	`for label_name in label_names:`
`766`	`766`	`label_data: dict[str, Any] = {`
`767`	`767`	`"name": label_name,`
`768`		`- "color": "#00" + "".join([random.choice("0123456789ABCDEF") for j in range(6)]), # noqa: S311`
	`768`	`+ "color": "#00" + "".join([random.choice("0123456789ABCDEF") for j in range(6)]), # noqa: S311 # nosec: B311`
`769`	`769`	`}`
`770`	`770`	`label_datas.append(label_data)`
`771`	`771`