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

pyproject.toml

@@ -78,6 +78,7 @@ darglint = """
     -not -path "./src/dsp_tools/commands/xmlupload/models/*" \
     -not -path "./src/dsp_tools/commands/project/models/*" \
     -not -path "./.git/*" \
+    -not -path "./.venv/*" \
     | xargs poetry run darglint -v 2\
 """
 clean = """
@@ -160,6 +161,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 +181,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 +201,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

@@ -146,7 +146,7 @@ def _convert_rows_to_xml(
 
         # this is a property-row
         else:
-            assert check_notna(row["prop name"])
+            assert check_notna(row["prop name"])  # noqa: S101 (use of assert in production code)
             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."

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

@@ -93,6 +93,9 @@ def stash_circular_references(
 
     Returns:
         stash: an object that contains the stashed references
+
+    Raises:
+        ValueError: If a link property of one of the resources is not "text" or "resptr"
     """
     stashed_standoff_values: list[StandoffStashItem] = []
     stashed_link_values: list[LinkValueStashItem] = []
@@ -101,13 +104,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

@@ -39,8 +39,8 @@ def tearDownClass(cls) -> None:
 
     def test_validate_lists_section_with_schema(self) -> None:
         """Test if the resource file 'src/dsp_tools/resources/schema/lists-only.json' can be accessed."""
-        cmd = f"dsp-tools create --lists-only --validate-only {self.test_project_systematic_file.absolute()}"
-        self._make_cli_call(cmd)
+        args = ["create", "--lists-only", "--validate-only", str(self.test_project_systematic_file.absolute())]
+        self._make_cli_call(args)
 
     def test_excel_to_json_resources(self) -> None:
         """
@@ -49,7 +49,7 @@ def test_excel_to_json_resources(self) -> None:
         """
         excel_file = Path("testdata/excel2json/excel2json_files/test-name (test_label)/resources.xlsx")
         out_file = self.testdata_tmp / "_out_resources.json"
-        self._make_cli_call(f"dsp-tools excel2resources '{excel_file.absolute()}' {out_file.absolute()}")
+        self._make_cli_call(["excel2resources", str(excel_file.absolute()), str(out_file.absolute())])
         out_file.unlink()
 
     def test_excel_to_json_properties(self) -> None:
@@ -59,12 +59,12 @@ def test_excel_to_json_properties(self) -> None:
         """
         excel_file = Path("testdata/excel2json/excel2json_files/test-name (test_label)/properties.xlsx")
         out_file = self.testdata_tmp / "_out_properties.json"
-        self._make_cli_call(f"dsp-tools excel2properties '{excel_file.absolute()}' {out_file.absolute()}")
+        self._make_cli_call(["excel2properties", str(excel_file.absolute()), str(out_file.absolute())])
         out_file.unlink()
 
     def test_validate_project(self) -> None:
         """Test if the resource file 'src/dsp_tools/resources/schema/project.json' can be accessed."""
-        self._make_cli_call(cli_call=f"dsp-tools create --validate-only {self.test_project_minimal_file.absolute()}")
+        self._make_cli_call(["create", "--validate-only", str(self.test_project_minimal_file.absolute())])
 
     def test_xml_upload(self) -> None:
         """Test if the resource file 'src/dsp_tools/resources/schema/data.xsd' can be accessed."""
@@ -77,37 +77,34 @@ def test_xml_upload(self) -> None:
             password=self.password,
             verbose=True,
         )
-        self._make_cli_call(f"dsp-tools xmlupload -v {self.test_data_minimal_file.absolute()}")
+        self._make_cli_call(["xmlupload", "-v", str(self.test_data_minimal_file.absolute())])
 
     def _make_cli_call(
         self,
-        cli_call: str,
+        args: list[str],
         working_directory: Path = cwd,
     ) -> None:
         """
-        Execute a CLI call, capture its stdout and stderr,
+        Execute a CLI call to dsp-tools, capture its stdout and stderr,
         and raise an AssertionError with stdout and stderr if it fails.
-        Before every call, preprend "poetry run",
-        so that the correct virtual environment is used.
 
         Args:
-            cli_call: a command that can be executed by the system shell
+            args: list of arguments for the dsp-tools CLI call
             working_directory: working directory where to execute the call. Defaults to the class' default_cwd.
 
         Raises:
             AssertionError: detailed diagnostic message with stdout and stderr if the call fails
         """
         try:
             subprocess.run(
-                f"poetry run {cli_call}",
+                ["poetry", "run", "dsp-tools", *args],  # noqa: S607 (Starting a process with a partial executable path)
                 check=True,
-                shell=True,
                 capture_output=True,
                 cwd=working_directory,
             )
         except subprocess.CalledProcessError as e:
             msg = (
-                f"Failed CLI call:\n'{cli_call}'\n\n"
+                f"Failed CLI call:\n'{args}'\n\n"
                 f"Stdout:\n{e.output.decode('utf-8')}\n\n"
                 f"Stderr:\n{e.stderr.decode('utf-8')}"
             )