chore: replace bandit action by ruff's bandit ruleset (DEV-3026)

pyproject.toml

@@ -160,6 +160,7 @@ select = [
     "A",        # flake8-builtins
     "BLE",      # flake8-blind-except
     "ARG",      # flake8-unused-arguments
+    "S",        # flake8-bandit plugin which checks for security issues
     "YTT",      # flake8-2020 plugin, which checks for misuse of `sys.version` or `sys.version_info`
     "ASYNC",    # flake8-async plugin, which checks for bad async / asyncio practices
     "ISC",      # flake8-implicit-str-concat plugin, which checks for problematic string concatenation
@@ -179,6 +180,10 @@ select = [
 ]
 ignore = [
     "ISC001",   # flake8-implicit-str-concat: single-line-implicit-string-concatenation  # incompatible with the formatter
+    "S105",     # flake8-bandit: hardcoded password
+    "S106",     # flake8-bandit: hardcoded password
+    "S603",     # flake8-bandit: subprocess-without-shell-equals-true
+    "S320",     # flake8-bandit: xml parsing vulnerable to XML attacks
 ]
 
 
@@ -195,6 +200,7 @@ ignore = [
     "D103",     # pydocstyle: undocumented-public-function
     "D102",     # pydocstyle: undocumented-public-method
     "D101",     # pydocstyle: undocumented-public-class
+    "S101",     # flake8-bandit: use of assert
 ]
 
 

src/dsp_tools/cli/entry_point.py

@@ -118,8 +118,8 @@ def _parse_arguments(
 
 
 def _get_version() -> str:
-    result = subprocess.run("pip freeze | grep dsp-tools", check=False, shell=True, capture_output=True)
-    _detail_version = result.stdout.decode("utf-8")
+    result = subprocess.run("pip freeze".split(), check=False, capture_output=True).stdout.decode("utf-8")
+    _detail_version = next(x for x in result.split("\n") if "dsp-tools" in x)
     # _detail_version has one of the following formats:
     # - 'dsp-tools==5.0.3\n'
     # - 'dsp-tools==5.6.0.post9\n'

src/dsp_tools/commands/excel2xml/excel2xml_cli.py

@@ -127,26 +127,17 @@ def _convert_rows_to_xml(
 
     for index, row in dataframe.iterrows():
         row_number = int(str(index)) + 2
-        # either the row is a resource-row or a property-row, but not both
-        if check_notna(row["id"]) == check_notna(row["prop name"]):
-            raise BaseError(
-                f"Exactly 1 of the 2 columns 'id' and 'prop name' must be filled. "
-                f"Excel row {row_number} has too many/too less entries:\n"
-                f"id:        '{row['id']}'\n"
-                f"prop name: '{row['prop name']}'"
-            )
 
-        # this is a resource-row
-        elif check_notna(row["id"]):
+        if check_notna(row["id"]):
+            # this is a resource-row
             # the previous resource is finished, a new resource begins: append the previous to the resulting list
             # in all cases (except for the very first iteration), a previous resource exists
             if resource is not None:
                 resources.append(resource)
             resource = _convert_resource_row_to_xml(row_number=row_number, row=row)
 
-        # this is a property-row
-        else:
-            assert check_notna(row["prop name"])
+        elif check_notna(row["prop name"]):
+            # this is a property-row
             if resource is None:
                 raise BaseError(
                     "The first row of your Excel/CSV is invalid. The first row must define a resource, not a property."
@@ -159,6 +150,15 @@ def _convert_rows_to_xml(
             )
             resource.append(prop)
 
+        else:
+            # either the row is a resource-row or a property-row, but not both
+            raise BaseError(
+                f"Exactly 1 of the 2 columns 'id' and 'prop name' must be filled. "
+                f"Excel row {row_number} has too many/too less entries:\n"
+                f"id:        '{row['id']}'\n"
+                f"prop name: '{row['prop name']}'"
+            )
+
     # append the resource of the very last iteration of the for loop
     if resource is not None:
         resources.append(resource)

src/dsp_tools/commands/excel2xml/excel2xml_lib.py

@@ -876,8 +876,16 @@ def make_geometry_prop(
     for val in values:
         try:
             value_as_dict = json.loads(str(val.value))
-            assert value_as_dict["type"] in ["rectangle", "circle", "polygon"]
-            assert isinstance(value_as_dict["points"], list)
+            if value_as_dict["type"] not in ["rectangle", "circle", "polygon"]:
+                raise BaseError(
+                    f"Failed validation in resource '{calling_resource}', property '{name}': "
+                    f"The 'type' of the JSON geometry object must be 'rectangle', 'circle', or 'polygon'."
+                )
+            if not isinstance(value_as_dict["points"], list):
+                raise BaseError(
+                    f"Failed validation in resource '{calling_resource}', property '{name}': "
+                    f"The 'points'of the JSON geometry object must be a list of points."
+                )
         except (json.JSONDecodeError, TypeError, IndexError, KeyError, AssertionError):
             raise BaseError(
                 f"Failed validation in resource '{calling_resource}', property '{name}': "

src/dsp_tools/commands/fast_xmlupload/upload_files.py

@@ -66,7 +66,7 @@ def _get_paths_from_pkl_files(pkl_files: list[Path]) -> list[Path]:
     """
     orig_paths_2_processed_paths: list[tuple[Path, Optional[Path]]] = []
     for pkl_file in pkl_files:
-        orig_paths_2_processed_paths.extend(pickle.loads(pkl_file.read_bytes()))
+        orig_paths_2_processed_paths.extend(pickle.loads(pkl_file.read_bytes()))  # noqa: S301 (deserialize untrusted data)
 
     processed_paths: list[Path] = []
     for orig_path, processed_path in orig_paths_2_processed_paths:

src/dsp_tools/commands/fast_xmlupload/upload_xml.py

@@ -29,7 +29,7 @@ def _get_paths_from_pkl_files(pkl_files: list[Path]) -> dict[str, str]:
     """
     orig_path_2_processed_path: list[tuple[Path, Optional[Path]]] = []
     for pkl_file in pkl_files:
-        orig_path_2_processed_path.extend(pickle.loads(pkl_file.read_bytes()))
+        orig_path_2_processed_path.extend(pickle.loads(pkl_file.read_bytes()))  # noqa: S301 (deserialize untrusted data)
 
     orig_path_2_uuid_filename: dict[str, str] = {}
     for orig_path, processed_path in orig_path_2_processed_path:

src/dsp_tools/commands/rosetta.py

@@ -21,7 +21,7 @@ def _update_possibly_existing_repo(rosetta_folder: Path) -> bool:
     is_rosetta_up_to_date = True
     if rosetta_folder.is_dir():
         print(f"Execute 'git pull' in {rosetta_folder}...")
-        completed_process = subprocess.run("git pull", shell=True, cwd=rosetta_folder, check=False)
+        completed_process = subprocess.run("git pull".split(), cwd=rosetta_folder, check=False)
         if not completed_process or completed_process.returncode != 0:
             print(f"'git pull' failed. Remove '{rosetta_folder}'...")
             shutil.rmtree(rosetta_folder, ignore_errors=True)
@@ -48,8 +48,7 @@ def _clone_repo(
     """
     print(f"Clone into {rosetta_folder}...")
     completed_process = subprocess.run(
-        "git clone https://github.com/dasch-swiss/082E-rosetta-scripts.git",
-        shell=True,
+        "git clone https://github.com/dasch-swiss/082E-rosetta-scripts.git".split(),
         cwd=enclosing_folder,
         check=False,
     )

src/dsp_tools/commands/start_stack.py

@@ -152,8 +152,7 @@ def _start_up_fuseki(self) -> None:
             UserError: if the database cannot be started
         """
         completed_process = subprocess.run(
-            "docker compose up db -d",
-            shell=True,
+            "docker compose up db -d".split(),
             cwd=self.__docker_path_of_user,
             check=False,
         )
@@ -256,12 +255,11 @@ def _start_remaining_docker_containers(self) -> None:
         """
         if self.__stack_configuration.latest_dev_version:
             subprocess.run(
-                "docker pull daschswiss/knora-api:latest",
-                shell=True,
+                "docker pull daschswiss/knora-api:latest".split(),
                 cwd=self.__docker_path_of_user,
                 check=True,
             )
-        subprocess.run("docker compose up -d", shell=True, cwd=self.__docker_path_of_user, check=True)
+        subprocess.run("docker compose up -d".split(), cwd=self.__docker_path_of_user, check=True)
         print("DSP-API is now running on http://0.0.0.0:3333/ and DSP-APP on http://0.0.0.0:4200/")
 
     def _execute_docker_system_prune(self) -> None:
@@ -281,7 +279,7 @@ def _execute_docker_system_prune(self) -> None:
                     "If you are unsure what that means, just type y and press Enter. [y/n]"
                 )
         if prune_docker == "y":
-            subprocess.run("docker system prune -f", shell=True, cwd=self.__docker_path_of_user, check=False)
+            subprocess.run("docker system prune -f".split(), cwd=self.__docker_path_of_user, check=False)
 
     def _start_docker_containers(self) -> None:
         """
@@ -320,5 +318,5 @@ def stop_stack(self) -> bool:
         Returns:
             True if everything went well, False otherwise
         """
-        subprocess.run("docker compose down --volumes", shell=True, cwd=self.__docker_path_of_user, check=True)
+        subprocess.run("docker compose down --volumes".split(), cwd=self.__docker_path_of_user, check=True)
         return True

src/dsp_tools/commands/xmlupload/stash/stash_circular_references.py

@@ -101,13 +101,14 @@ def stash_circular_references(
         if res.res_id not in stash_lookup:
             continue
         for link_prop in res.get_props_with_links():
-            assert link_prop.valtype in ["text", "resptr"]
             if link_prop.valtype == "text":
                 standoff_stash_item = _stash_standoff(res.res_id, res.restype, link_prop, stash_lookup)
                 stashed_standoff_values.extend(standoff_stash_item)
             elif link_prop.valtype == "resptr":
                 link_stash_item = _stash_resptr(res.res_id, res.restype, link_prop, stash_lookup, permission_lookup)
                 stashed_link_values.extend(link_stash_item)
+            else:
+                raise ValueError(f"Unknown value type: '{link_prop.valtype}' (should be 'text' or 'resptr')")
 
             if len(link_prop.values) == 0:
                 # if all values of a link property have been stashed, the property needs to be removed

src/dsp_tools/commands/xmlupload/upload_config.py

@@ -27,7 +27,7 @@ def _transform_server_url_to_foldername(server: str) -> str:
         r"^api\.": "",
         r":\d{2,5}/?$": "",
         r"/$": "",
-        r"0.0.0.0": "localhost",
+        r"0.0.0.0": "localhost",  # noqa: S104 (hardcoded-bind-all-interfaces)
     }
     for pattern, repl in server_substitutions.items():
         server = regex.sub(pattern, repl, server)

test/e2e/test_00A1_import_scripts.py

@@ -26,7 +26,7 @@ def generated_xml_file() -> Iterator[Path]:
 def test_script(generated_xml_file: Path) -> None:
     """Execute the import script in its directory"""
     # pull the latest state of the git submodule
-    subprocess.run("git submodule update --init --recursive", check=True, shell=True)
+    subprocess.run("git submodule update --init --recursive".split(), check=True)
     from dsp_tools.import_scripts import import_script
 
     # execute the import script in its directory

test/e2e/test_cli.py

@@ -99,9 +99,8 @@ def _make_cli_call(
         """
         try:
             subprocess.run(
-                f"poetry run {cli_call}",
+                f"poetry run {cli_call}".split(),
                 check=True,
-                shell=True,
                 capture_output=True,
                 cwd=working_directory,
             )