MAINT: library vulnerabilities action (#505)

* library vulnerabilities action

* fix workflow file

* address bandit issues with exclusions and code changes

* whitespace in pyproject.toml

* remove commented out import

* address some copilot review comments

Author: iboyd-ansys

.github/workflows/ci.yml

@@ -62,6 +62,19 @@ jobs:
           operating-system: ${{ matrix.os }}
           python-version: ${{ matrix.python-version }}
 
+  check-vulnerabilities:
+    name: "Check library vulnerabilities"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: ansys/actions/[email protected]
+        with:
+          python-version: ${{ env.MAIN_PYTHON_VERSION }}
+          token: ${{ secrets.PYANSYS_CI_BOT_TOKEN }}
+          python-package-name: ${{ env.PACKAGE_NAME }}
+          dev-mode: ${{ github.ref != 'refs/heads/main' }}
+          #upload-reports: True
+          #hide-log: false
+
   build:
     name: Build package, incl. API generation
     needs: [smoke-tests]

pyproject.toml

@@ -19,6 +19,7 @@ maintainers = [
 dependencies = [
     "ansys-api-systemcoupling==0.2.0",
     "ansys-platform-instancemanagement~=1.0",
+    "docker>=7.1.0",
     "grpcio>=1.30.0",
     "grpcio-status>=1.30.0",
     "googleapis-common-protos>=1.50.0",

src/ansys/systemcoupling/core/adaptor/impl/get_syc_version.py

@@ -44,11 +44,14 @@ def get_syc_version(api) -> str:
     def clean_version_string(version_in: str) -> str:
         year, _, release = version_in.partition(" ")
         if len(year) == 4 and year.startswith("20") and release.startswith("R"):
+            # Exclude Bandit check. The try-except-pass is only used to simplify logic.
+            # An exception will be thrown in any case, but it *also* gets thrown for
+            # input that does not match the above 'if' condition.
             try:
                 year = int(year[2:])
                 release = int(release[1:])
                 return f"{year}.{release}"
-            except:
+            except:  # nosec B110
                 pass
         raise RuntimeError(
             f"Version string {version_in} has invalid format (expect '20yy Rn')."

src/ansys/systemcoupling/core/adaptor/impl/injected_commands.py

@@ -187,7 +187,9 @@ def _ensure_file_available(session: SessionProtocol, filepath: str) -> str:
     file_name = os.path.basename(filepath)
     root_name, _, ext = file_name.rpartition(".")
     ext = f".{ext}" if ext else ""
-    new_name = f"{root_name}_{int(time.time())}_{random.randint(1, 10000000)}{ext}"
+    # Exclude Bandit check as random number is simply being used to create a unique
+    # file name, not for security/cryptographic purposes.
+    new_name = f"{root_name}_{int(time.time())}_{random.randint(1, 10000000)}{ext}"  # nosec B311
 
     session._native_api.ExecPythonString(
         PythonString=f"import shutil\nshutil.copy('{filepath}', '{new_name}')"

src/ansys/systemcoupling/core/charts/csv_chartdata.py

@@ -32,6 +32,7 @@
     TimestepData,
     TransferSeriesInfo,
 )
+from ansys.systemcoupling.core.util.assertion import assert_
 
 HeaderList = list[str]
 ChartData = list[list[float]]
@@ -231,8 +232,8 @@ def _parse_suffix(header: str, part_disp_name: str) -> str:
 
 def parse_csv_metadata(interface_name: str, headers: list[str]) -> InterfaceInfo:
     intf_info = InterfaceInfo(name=interface_name)
-    assert headers[0] == "Iteration"
-    assert headers[1] == "Step"
+    assert_(headers[0] == "Iteration", 'Header expected to be "Iteration"')
+    assert_(headers[1] == "Step", 'Header expected to be "Step"')
     intf_info.is_transient = headers[2] == "Time"
 
     start_index = 3 if intf_info.is_transient else 2
@@ -263,7 +264,7 @@ def parse_csv_metadata(interface_name: str, headers: list[str]) -> InterfaceInfo
 
             intf_disp_name = intf_or_part_disp_name
             if data_index == 0:
-                assert intf_info.display_name == ""
+                assert_(intf_info.display_name == "", "display_name should be empty")
                 intf_info.display_name = intf_disp_name
             series_info = TransferSeriesInfo(
                 data_index,

src/ansys/systemcoupling/core/charts/plot_functions.py

@@ -34,8 +34,12 @@
 
 
 def create_and_show_plot(spec: PlotSpec, csv_list: list[str]) -> Plotter:
-    assert len(spec.interfaces) == 1, "Plots currently only support one interface"
-    assert len(spec.interfaces) == len(csv_list)
+    if len(spec.interfaces) != 1:
+        raise ValueError("Plots currently only support one interface")
+    if len(spec.interfaces) != len(csv_list):
+        raise ValueError(
+            "'csv_list' should have length equal to the number of interfaces"
+        )
 
     manager = PlotDefinitionManager(spec)
     reader = CsvChartDataReader(spec.interfaces[0].name, csv_list[0])
@@ -61,8 +65,12 @@ def solve_with_live_plot(
     csv_list: list[str],
     solve_func: Callable[[], None],
 ):
-    assert len(spec.interfaces) == 1, "Plots currently only support one interface"
-    assert len(spec.interfaces) == len(csv_list)
+    if len(spec.interfaces) != 1:
+        raise ValueError("Plots currently only support one interface")
+    if len(spec.interfaces) != len(csv_list):
+        raise ValueError(
+            "'csv_list' should have length equal to the number of interfaces"
+        )
 
     manager = PlotDefinitionManager(spec)
     dispatcher = MessageDispatcher()

src/ansys/systemcoupling/core/charts/plotter.py

@@ -37,6 +37,7 @@
 from ansys.systemcoupling.core.charts.plotdefinition_manager import (
     PlotDefinitionManager,
 )
+from ansys.systemcoupling.core.util.assertion import assert_
 
 
 def _process_timestep_data(
@@ -282,13 +283,13 @@ def show_animated(self):
         # this (assume the wait function is stored as an
         # attribute):
         #
-        # assert self._wait_for_metadata is not None
+        # assert_(self._wait_for_metadata is not None)
         # metadata = self._wait_for_metadata()
         # if metadata is not None:
         #     self.set_metadata(metadata)
         # else:
         #     return
-        assert self._request_update is not None
+        assert_(self._request_update is not None)
 
         self.ani = FuncAnimation(
             self._fig,

src/ansys/systemcoupling/core/client/grpc_client.py

@@ -101,6 +101,7 @@ def _reset(self):
         self.__output_thread = None
         self.__pim_instance = None
         self.__skip_exit = False
+        self.__container = None
 
     @classmethod
     def _cleanup(cls):
@@ -161,7 +162,9 @@ def start_container_and_connect(
         """Start the System Coupling container and establish a connection."""
         LOG.debug("Starting container...")
         port = port if port is not None else _find_port()
-        start_container(mounted_from, mounted_to, network, port, version)
+        self.__container = start_container(
+            mounted_from, mounted_to, network, port, version
+        )
         LOG.debug("...started")
         self._connect(_LOCALHOST_IP, port)
 
@@ -307,12 +310,18 @@ def exit(self):
             try:
                 self.__ostream_service.end_streaming()
             except Exception as e:
-                LOG.debug("Exception on OutputStreamService.end_straming(): " + str(e))
+                LOG.debug(f"Exception on OutputStreamService.end_straming(): {e}")
             self.__process_service.quit()
             self.__channel = None
         if self.__process:
             self.__process.end()
             self.__process = None
+        if self.__container:
+            try:
+                self.__container.stop()
+            except Exception as e:
+                LOG.debug(f"Exception from container.stop(): {e}")
+            self.__container = None
         if self.__pim_instance is not None:
             self.__pim_instance.delete()
             self.__pim_instance = None

src/ansys/systemcoupling/core/client/syc_container.py

@@ -22,9 +22,9 @@
 
 import os
 from pathlib import Path
-import subprocess
 
 from ansys.systemcoupling.core.syc_version import SYC_VERSION_DOT, normalize_version
+from ansys.systemcoupling.core.util.logging import LOG
 
 _MPI_VERSION_VAR = "FLUENT_INTEL_MPI_VERSION"
 _MPI_VERSION = "2021"
@@ -41,15 +41,22 @@ def _image_tag(version: str) -> str:
 
 def start_container(
     mounted_from: str, mounted_to: str, network: str, port: int, version: str
-) -> None:
+) -> object:
     """Start a System Coupling container.
 
     Parameters
     ----------
     port : int
         gPRC server local port, mapped to the same port in container.
+
+    Returns
+    -------
+    object
+        The container instance (``Container`` object from Python docker library).
     """
-    args = ["-m", "cosimgui", f"--grpcport=0.0.0.0:{port}"]
+    import docker
+
+    LOG.debug("Starting System Coupling docker container...")
 
     if version:
         image_tag = _image_tag(version)
@@ -58,47 +65,45 @@ def start_container(
 
     mounted_from = str(Path(mounted_from).absolute())
 
-    run_args = [
-        "docker",
-        "run",
-        "-d",
-        "--rm",
-        "-p",
-        f"{port}:{port}",
-        "-v",
-        f"{mounted_from}:{mounted_to}",
-        "-w",
-        mounted_to,
-        "-e",
-        f"{_MPI_VERSION_VAR}={_MPI_VERSION}",
-        "-e",
-        f"AWP_ROOT=/ansys_inc",
-        f"ghcr.io/ansys/pysystem-coupling:{image_tag}",
-    ] + args
+    image_name = f"ghcr.io/ansys/pysystem-coupling:{image_tag}"
+
+    environment = [f"{_MPI_VERSION_VAR}={_MPI_VERSION}", f"AWP_ROOT=/ansys_inc"]
 
+    # Additional environment
     container_user = os.getenv("SYC_CONTAINER_USER")
     if container_user:
-        idx = run_args.index("-p")
-        run_args.insert(idx, container_user)
-        run_args.insert(idx, "--user")
         # Licensing can't log to default location if user is not the default 'root'
-        run_args.insert(idx, f"ANSYSLC_APPLOGDIR={mounted_to}")
-        run_args.insert(idx, "-e")
+        environment.append(f"ANSYSLC_APPLOGDIR={mounted_to}")
+        # See also "user" argument added to args below
 
     license_server = os.getenv("ANSYSLMD_LICENSE_FILE")
     if license_server:
-        idx = run_args.index("-e")
-        run_args.insert(idx, f"ANSYSLMD_LICENSE_FILE={license_server}")
-        run_args.insert(idx, "-e")
-
+        environment.append(f"ANSYSLMD_LICENSE_FILE={license_server}")
+
+    run_args = dict(
+        image=image_name,
+        command=["-m", "cosimgui", f"--grpcport=0.0.0.0:{port}"],
+        detach=True,
+        environment=environment,
+        remove=True,
+        ports={f"{port}/tcp": f"{port}"},
+        volumes=[f"{mounted_from}:{mounted_to}"],
+        working_dir=f"{mounted_to}",
+    )
+
+    # Additional args
+    if container_user:
+        run_args["user"] = container_user
     if network:
-        idx = run_args.index("-p")
-        run_args.insert(idx, network)
-        run_args.insert(idx, "--network")
+        run_args["network"] = network
 
-    subprocess.run(run_args)
-    return port
+    docker_client = docker.from_env()
+    return docker_client.containers.run(**run_args)
 
 
 def create_network(name):
-    subprocess.run(["docker", "network", "create", name])
+    import docker
+
+    LOG.debug(f"Creating docker network '{name}'")
+    docker_client = docker.from_env()
+    docker_client.networks.create(name)

src/ansys/systemcoupling/core/client/syc_process.py

@@ -23,7 +23,9 @@
 from copy import deepcopy
 import os
 import platform
-import subprocess
+
+# Exclude Bandit check. Subprocess is needed to start the System Coupling.
+import subprocess  # nosec B404
 
 import psutil
 
@@ -99,7 +101,8 @@ def _start_system_coupling(
         args += extra_args
 
     LOG.info(f"Starting System Coupling: {args[0]}")
-    return subprocess.Popen(
+    # Exclude Bandit check. No untrusted input to arguments.
+    return subprocess.Popen(  # nosec B603
         args,
         env=env,
         cwd=working_dir,