Use shell=False for Popen on Windows

pytensor/link/c/cmodule.py

@@ -10,6 +10,7 @@
 import pickle
 import platform
 import re
+import shlex
 import shutil
 import stat
 import subprocess
@@ -2104,49 +2105,48 @@
             )
             detect_march = False
 
-        if detect_march:
-            GCC_compiler.march_flags = []
+        def get_lines(cmd: list[str], parse: bool = True) -> list[str] | None:
+            p = subprocess_Popen(
+                cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE,
+            )
+            # For mingw64 with GCC >= 4.7, passing os.devnull
+            # as stdin (which is the default) results in the process
+            # waiting forever without returning. For that reason,
+            # we use a pipe, and use the empty string as input.
+            (stdout, stderr) = p.communicate(input=b"")
+            if p.returncode != 0:
+                return None
 
-            def get_lines(cmd, parse=True):
-                p = subprocess_Popen(
-                    cmd,
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
-                    stdin=subprocess.PIPE,
-                    shell=True,
-                )
-                # For mingw64 with GCC >= 4.7, passing os.devnull
-                # as stdin (which is the default) results in the process
-                # waiting forever without returning. For that reason,
-                # we use a pipe, and use the empty string as input.
-                (stdout, stderr) = p.communicate(input=b"")
-                if p.returncode != 0:
-                    return None
-
-                lines = BytesIO(stdout + stderr).readlines()
-                lines = (l.decode() for l in lines)
-                if parse:
-                    selected_lines = []
-                    for line in lines:
-                        if (
-                            "COLLECT_GCC_OPTIONS=" in line
-                            or "CFLAGS=" in line
-                            or "CXXFLAGS=" in line
-                            or "-march=native" in line
-                        ):
-                            continue
-                        selected_lines.extend(
-                            line.strip()
-                            for reg in ("-march=", "-mtune=", "-target-cpu", "-mabi=")
-                            if reg in line
-                        )
-                    lines = list(set(selected_lines))  # to remove duplicate
+            lines_bytes = BytesIO(stdout + stderr).readlines()
+            lines = [l.decode() for l in lines_bytes]
+            if parse:
+                selected_lines: list[str] = []
+                for line in lines:
+                    if (
+                        "COLLECT_GCC_OPTIONS=" in line
+                        or "CFLAGS=" in line
+                        or "CXXFLAGS=" in line
+                        or "-march=native" in line
+                    ):
+                        continue
+                    selected_lines.extend(
+                        line.strip()
+                        for reg in ("-march=", "-mtune=", "-target-cpu", "-mabi=")
+                        if reg in line
+                    )
+                lines = list(set(selected_lines))  # to remove duplicate
 
-                return lines
+            return lines
+
+        if detect_march:
+            GCC_compiler.march_flags = []
 
             # The '-' at the end is needed. Otherwise, g++ do not output
             # enough information.
-            native_lines = get_lines(f"{config.cxx} -march=native -E -v -")
+            native_lines = get_lines([config.cxx, "-march=native", "-E", "-v", "-"])
             if native_lines is None:
                 _logger.info(
                     "Call to 'g++ -march=native' failed, not setting -march flag"
@@ -2161,7 +2161,7 @@
                     # That means we did not select the right lines, so
                     # we have to report all the lines instead
                     reported_lines = get_lines(
-                        f"{config.cxx} -march=native -E -v -", parse=False
+                        [config.cxx, "-march=native", "-E", "-v", "-"], parse=False
                     )
                 else:
                     reported_lines = native_lines
@@ -2174,10 +2174,12 @@
                     f" problem:\n {reported_lines}"
                 )
             else:
-                default_lines = get_lines(f"{config.cxx} -E -v -")
+                default_lines = get_lines([config.cxx, "-E", "-v", "-"])
                 _logger.info(f"g++ default lines: {default_lines}")
                 if len(default_lines) < 1:
-                    reported_lines = get_lines(f"{config.cxx} -E -v -", parse=False)
+                    reported_lines = get_lines(
+                        [config.cxx, "-E", "-v", "-"], parse=False
+                    )
                     warnings.warn(
                         "PyTensor was not able to find the "
                         "default g++ parameters. This is needed to tune "
@@ -2610,15 +2612,15 @@
         cmd.append(f"{path_wrapper}{cppfilename}{path_wrapper}")
         cmd.extend(GCC_compiler.linking_patch(lib_dirs, libs))
         # print >> sys.stderr, 'COMPILING W CMD', cmd
-        _logger.debug(f"Running cmd: {' '.join(cmd)}")
+        _logger.debug(f"Running cmd: {shlex.join(cmd)}")
 
         def print_command_line_error():
             # Print command line when a problem occurred.
             print(
                 ("Problem occurred during compilation with the command line below:"),
                 file=sys.stderr,
             )
-            print(" ".join(cmd), file=sys.stderr)
+            print(shlex.join(cmd), file=sys.stderr)
 
         try:
             p_out = output_subprocess_Popen(cmd)
@@ -2777,13 +2779,12 @@
         return libs
 
     def get_cxx_library_dirs():
-        cmd = f"{config.cxx} -print-search-dirs"
+        cmd = [config.cxx, "-print-search-dirs"]
         p = subprocess_Popen(
             cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             stdin=subprocess.PIPE,
-            shell=True,
         )
         (stdout, stderr) = p.communicate(input=b"")
         if p.returncode != 0:

pytensor/utils.py

@@ -123,7 +123,7 @@ def maybe_add_to_os_environ_pathlist(var: str, newpath: Path | str) -> None:
         pass
 
 
-def subprocess_Popen(command: str | list[str], **params):
+def subprocess_Popen(command: list[str], **params) -> subprocess.Popen:
     """
     Utility function to work around windows behavior that open windows.
 
@@ -137,37 +137,17 @@ def subprocess_Popen(command: str | list[str], **params):
         except AttributeError:
             startupinfo.dwFlags |= subprocess._subprocess.STARTF_USESHOWWINDOW  # type: ignore[attr-defined]
 
-        # Anaconda for Windows does not always provide .exe files
-        # in the PATH, they also have .bat files that call the corresponding
-        # executable. For instance, "g++.bat" is in the PATH, not "g++.exe"
-        # Unless "shell=True", "g++.bat" is not executed when trying to
-        # execute "g++" without extensions.
-        # (Executing "g++.bat" explicitly would also work.)
-        params["shell"] = True
         # "If shell is True, it is recommended to pass args as a string rather than as a sequence." (cite taken from https://docs.python.org/2/library/subprocess.html#frequently-used-arguments)
         # In case when command arguments have spaces, passing a command as a list will result in incorrect arguments break down, and consequently
         # in "The filename, directory name, or volume label syntax is incorrect" error message.
         # Passing the command as a single string solves this problem.
         if isinstance(command, list):
-            command = " ".join(command)
+            command = " ".join(command)  # type: ignore[assignment]
 
-    # Using the dummy file descriptors below is a workaround for a
-    # crash experienced in an unusual Python 2.4.4 Windows environment
-    # with the default None values.
-    stdin = None
-    if "stdin" not in params:
-        stdin = Path(os.devnull).open()
-        params["stdin"] = stdin.fileno()
-
-    try:
-        proc = subprocess.Popen(command, startupinfo=startupinfo, **params)
-    finally:
-        if stdin is not None:
-            stdin.close()
-    return proc
+    return subprocess.Popen(command, startupinfo=startupinfo, **params)
 
 
-def call_subprocess_Popen(command, **params):
+def call_subprocess_Popen(command: list[str], **params) -> int:
     """
     Calls subprocess_Popen and discards the output, returning only the
     exit code.
@@ -185,13 +165,17 @@ def call_subprocess_Popen(command, **params):
     return returncode
 
 
-def output_subprocess_Popen(command, **params):
+def output_subprocess_Popen(command: list[str], **params) -> tuple[bytes, bytes, int]:
     """
     Calls subprocess_Popen, returning the output, error and exit code
     in a tuple.
     """
     if "stdout" in params or "stderr" in params:
         raise TypeError("don't use stderr or stdout with output_subprocess_Popen")
+    if "encoding" in params:
+        raise TypeError(
+            "adjust the output_subprocess_Popen type annotation to support str"
+        )
     params["stdout"] = subprocess.PIPE
     params["stderr"] = subprocess.PIPE
     p = subprocess_Popen(command, **params)

tests/compile/function/test_function.py

@@ -1,5 +1,4 @@
 import pickle
-import re
 import shutil
 import tempfile
 from pathlib import Path
@@ -50,8 +49,7 @@ def test_function_name():
     x = vector("x")
     func = function([x], x + 1.0)
 
-    regex = re.compile(f".*{__file__}c?")
-    assert regex.match(func.name) is not None
+    assert __file__ in func.name
 
 
 def test_trust_input():