REFACTOR: Improve code quality and handling of subprocess calls (#5995)

Co-authored-by: pyansys-ci-bot <92810346+pyansys-ci-bot@users.noreply.github.com>
Co-authored-by: Maxime Rey <87315832+MaxJPRey@users.noreply.github.com>

Author: 3 people

doc/changelog.d/5995.miscellaneous.md

@@ -0,0 +1 @@
+Improve code quality and handling of subprocess calls

src/ansys/aedt/core/application/analysis.py

@@ -60,6 +60,7 @@
 from ansys.aedt.core.generic.numbers import decompose_variable_value
 from ansys.aedt.core.generic.numbers import is_number
 from ansys.aedt.core.generic.settings import settings
+from ansys.aedt.core.internal.errors import AEDTRuntimeError
 from ansys.aedt.core.modules.boundary.layout_boundary import NativeComponentObject
 from ansys.aedt.core.modules.boundary.layout_boundary import NativeComponentPCB
 from ansys.aedt.core.modules.design_xploration import OptimizationSetups
@@ -1903,6 +1904,7 @@ def stop_simulations(self, clean_stop=True):
         """
         return self.desktop_class.stop_simulations(clean_stop=clean_stop)
 
+    # flake8: noqa: E501
     @pyaedt_function_handler(filename="file_name", numcores="cores", num_tasks="tasks", setup_name="setup")
     def solve_in_batch(
         self,
@@ -1919,6 +1921,11 @@ def solve_in_batch(
         .. note::
            To use this function, the project must be closed.
 
+        .. warning::
+            Do not execute this function with untrusted input parameters.
+            See the :ref:`security guide<https://aedt.docs.pyansys.com/version/stable/User_guide/security_consideration.html>`
+            for details.
+
         Parameters
         ----------
         file_name : str, optional
@@ -1945,6 +1952,15 @@ def solve_in_batch(
          bool
            ``True`` when successful, ``False`` when failed.
         """
+        try:
+            cores = int(cores)
+        except ValueError:
+            raise ValueError(f"The number of cores is not a valid integer.")
+        try:
+            tasks = int(tasks)
+        except ValueError:
+            raise ValueError(f"The number of tasks is not a valid integer.")
+
         inst_dir = self.desktop_install_dir
         self.last_run_log = ""
         self.last_run_job = ""
@@ -1979,46 +1995,37 @@ def solve_in_batch(
         if setup and design_name:
             options.append(f'{design_name}:{"Nominal" if setup in self.setup_names else "Optimetrics"}:{setup}')
         if is_linux and not settings.use_lsf_scheduler:
-            batch_run = [inst_dir + "/ansysedt"]
+            command = [inst_dir + "/ansysedt"]
         elif is_linux and settings.use_lsf_scheduler:  # pragma: no cover
+            if not isinstance(settings.lsf_ram, int) or settings.lsf_ram <= 0:
+                raise AEDTRuntimeError("Invalid memory value.")
+            if not settings.lsf_aedt_command:
+                raise AEDTRuntimeError("Invalid LSF AEDT command.")
+            command = [
+                "bsub",
+                "-n",
+                str(cores),
+                "-R",
+                f"span[ptile={cores}]",
+                "-R",
+                f"rusage[mem={settings.lsf_ram}]",
+                settings.lsf_aedt_command,
+            ]
             if settings.lsf_queue:
-                batch_run = [
-                    "bsub",
-                    "-n",
-                    str(cores),
-                    "-R",
-                    f"span[ptile={cores}]",
-                    "-R",
-                    f"rusage[mem={settings.lsf_ram}]",
-                    f"-queue {settings.lsf_queue}",
-                    settings.lsf_aedt_command,
-                ]
-            else:
-                batch_run = [
-                    "bsub",
-                    "-n",
-                    str(cores),
-                    "-R",
-                    f"span[ptile={cores}]",
-                    "-R",
-                    f"rusage[mem={settings.lsf_ram}]",
-                    settings.lsf_aedt_command,
-                ]
+                command.extend(["-queue", settings.lsf_queue])
         else:
-            batch_run = [inst_dir + "/ansysedt.exe"]
-        batch_run.extend(options)
-        batch_run.append(file_name)
+            command = [inst_dir + "/ansysedt.exe"]
+        command.extend(options)
+        command.append(file_name)
 
         # check for existing solution directory and delete it if it exists so we
         # don't have old .asol files etc
-
         self.logger.info("Solving model in batch mode on " + machine)
         if run_in_thread and is_windows:
-            DETACHED_PROCESS = 0x00000008
-            subprocess.Popen(batch_run, creationflags=DETACHED_PROCESS)
+            subprocess.Popen(command, creationflags=subprocess.DETACHED_PROCESS)  # nosec
             self.logger.info("Batch job launched.")
         else:
-            subprocess.Popen(batch_run)
+            subprocess.Popen(command)  # nosec
             self.logger.info("Batch job finished.")
 
         if machine == "localhost":

src/ansys/aedt/core/common_rpc.py

@@ -25,6 +25,7 @@
 import os
 from pathlib import Path
 import signal
+import socket
 import sys
 import tempfile
 import time
@@ -187,7 +188,11 @@ def pyaedt_service_manager(port=17878, aedt_version=None, student_version=False)
     os.environ["PYAEDT_SERVER_AEDT_NG"] = "True"
     os.environ["ANS_NODEPCHECK"] = str(1)
 
-    hostname = "0.0.0.0"
+    default_host = socket.gethostname()
+    hostname = os.getenv("AEDT_HOST", default_host)
+    if hostname == "0.0.0.0":  # nosec
+        logger.warning("The service is exposed on all network interfaces. This is a security risk.")
+
     t = ThreadedServer(
         ServiceManager,
         hostname=hostname,
@@ -248,7 +253,11 @@ def launch_server(port=18000, ansysem_path=None, non_graphical=False, threaded=T
     os.environ["ANS_NO_MONO_CLEANUP"] = str(1)
     os.environ["ANS_NODEPCHECK"] = str(1)
 
-    hostname = "0.0.0.0"
+    default_host = socket.gethostname()
+    hostname = os.getenv("AEDT_HOST", default_host)
+    if hostname == "0.0.0.0":  # nosec
+        logger.warning("The service is exposed on all network interfaces. This is a security risk.")
+
     if threaded:
         service = ThreadedServer
     else:

src/ansys/aedt/core/generic/data_handlers.py

@@ -294,24 +294,13 @@ def unique_string_list(element_list, only_string=True):
     -------
 
     """
-    if element_list:
-        if isinstance(element_list, list):
-            element_list = set(element_list)
-        elif isinstance(element_list, str):
-            element_list = [element_list]
-        else:
-            error_message = "Invalid list data"
-            try:
-                error_message += f" {element_list}"
-            except Exception:
-                pass
-            raise Exception(error_message)
+    if isinstance(element_list, str):
+        element_list = [element_list]
 
-        if only_string:
-            non_string_entries = [x for x in element_list if not isinstance(x, str)]
-            assert not non_string_entries, f"Invalid list entries {non_string_entries} are not a string!"
+    if only_string and any(not isinstance(x, str) for x in element_list):
+        raise TypeError("Invalid list entries, some elements are not of type string.")
 
-    return element_list
+    return list(set(element_list))
 
 
 @pyaedt_function_handler()
@@ -330,10 +319,10 @@ def string_list(element_list):
     list
 
     """
+    if not isinstance(element_list, (str, list)):
+        raise TypeError("Input must be a list or a string")
     if isinstance(element_list, str):
         element_list = [element_list]
-    else:
-        assert isinstance(element_list, str), "Input must be a list or a string"
     return element_list
 
 

src/ansys/aedt/core/internal/aedt_versions.py

@@ -31,6 +31,8 @@
 import os
 import warnings
 
+from ansys.aedt.core.generic import settings
+
 CURRENT_STABLE_AEDT_VERSION = 2025.1
 
 
@@ -106,7 +108,8 @@ def installed_versions(self):
                         v_key = f"20{version}.{release}"
                     return_dict[v_key] = os.environ[version_env_var]
                 except Exception:  # pragma: no cover
-                    pass
+                    if settings.logger:
+                        settings.logger.debug(f"Failed to parse version and release from {current_version_id}")
             self._installed_versions = return_dict
         return self._installed_versions
 

src/ansys/aedt/core/internal/grpc_plugin_dll_class.py

@@ -30,7 +30,6 @@
 from ctypes import py_object
 import os
 import re
-import socket
 import types
 
 from ansys.aedt.core.generic.general_methods import _retry_ntimes
@@ -219,18 +218,18 @@ def __getattr__(self, attrName):
                 return self.GetPropValue(propMap[attrName])
             raise GrpcApiError(f"Failed to retrieve attribute {attrName} from gRPC API")
 
-    def __setattr__(self, attrName, val):
-        if attrName in self.__dict__:
-            self.__dict__[attrName] = val
+    def __setattr__(self, attr, val):
+        if attr in self.__dict__:
+            self.__dict__[attr] = val
             return
         propMap = self.__GetPropAttributes()
         try:
-            if attrName in propMap:
-                self.SetPropValue(propMap[attrName], val)
+            if attr in propMap:
+                self.SetPropValue(propMap[attr], val)
                 return
         except Exception:
-            pass
-        super().__setattr__(attrName, val)
+            settings.logger.debug(f"Failed to update attribute {attr} of AedtPropServer instance.")
+        super().__setattr__(attr, val)
 
     def GetName(self):
         return self.__Invoke__("GetName", ())
@@ -331,21 +330,6 @@ def SetPyObjCalbacks(self):
         self.callbackGetObjID = RetObj_InObj_Func_type(self.GetAedtObjId)
         self.AedtAPI.SetPyObjCalbacks(self.callbackToCreateObj, self.callbackCreateBlock, self.callbackGetObjID)
 
-    @staticmethod
-    def _is_port_occupied(port, machine_name=""):
-        s = socket.socket()
-        try:
-            if not machine_name:
-                machine_name = "127.0.0.1"
-            s.connect((machine_name, port))
-        except socket.error:
-            success = False
-        else:
-            success = True
-        finally:
-            s.close()
-        return success
-
     def CreateAedtApplication(self, machine="", port=0, NGmode=False, alwaysNew=True):
         if not alwaysNew and port:
             grpc_channel = grpc.insecure_channel(f"{machine}:{port}")

src/ansys/aedt/core/workflows/hfss/move_it.py

@@ -28,6 +28,7 @@
 
 import ansys.aedt.core
 from ansys.aedt.core import get_pyaedt_app
+from ansys.aedt.core.internal.errors import AEDTRuntimeError
 from ansys.aedt.core.workflows.misc import get_aedt_version
 from ansys.aedt.core.workflows.misc import get_arguments
 from ansys.aedt.core.workflows.misc import get_port
@@ -369,7 +370,9 @@ def main(extension_args):
         cs.props["YAxisXvec"] = "0mm"
         cs.props["YAxisYvec"] = "0mm"
         cs.props["YAxisZvec"] = "1mm"
-        assert cs.update()
+        res = cs.update()
+        if not res:
+            raise AEDTRuntimeError("Failed to update the coordinate system.")
 
     else:
         cs = hfss.modeler.create_coordinate_system(
@@ -389,8 +392,9 @@ def main(extension_args):
         cs_rotated.props["YAxisYvec"] = "0"
         cs_rotated.props["YAxisZvec"] = "-1"
         cs_rotated.props["Reference CS"] = cs.name
-        assert cs_rotated.update()
-
+        res = cs_rotated.update()
+        if not res:
+            raise AEDTRuntimeError("Failed to update the coordinate system.")
     else:
         _ = hfss.modeler.create_coordinate_system(name=cs_name_rotated, reference_cs=cs.name, y_pointing=[0, 0, -1])