Run a command as a subprocess, capturing stderr and stdout. ``env_overrides`` keys map to a string (set/replace) or ``None`` (delete).
(
cmd: Sequence[str | Path],
capture_stdout: bool = True,
capture_stderr: bool = True,
log_cmd_str: str | None = None,
log_stdout: bool = True,
log_stderr: bool = True,
run_dir: str | None = None,
env_overrides: dict[str, str | None] | None = None,
)
| 156 | |
| 157 | |
| 158 | def run_subprocess( |
| 159 | cmd: Sequence[str | Path], |
| 160 | capture_stdout: bool = True, |
| 161 | capture_stderr: bool = True, |
| 162 | log_cmd_str: str | None = None, |
| 163 | log_stdout: bool = True, |
| 164 | log_stderr: bool = True, |
| 165 | run_dir: str | None = None, |
| 166 | env_overrides: dict[str, str | None] | None = None, |
| 167 | ) -> "subprocess.CompletedProcess[str]": |
| 168 | """Run a command as a subprocess, capturing stderr and stdout. |
| 169 | |
| 170 | ``env_overrides`` keys map to a string (set/replace) or ``None`` (delete). |
| 171 | """ |
| 172 | env = dict(os.environ) |
| 173 | env = _fix_subprocess_env(env) |
| 174 | for key, value in (env_overrides or {}).items(): |
| 175 | if value is None: |
| 176 | env.pop(key, None) |
| 177 | else: |
| 178 | env[key] = value |
| 179 | |
| 180 | if log_cmd_str is None: |
| 181 | log_cmd_str = " ".join(str(c) for c in cmd) |
| 182 | _LOGGER.info(f"running {log_cmd_str}") |
| 183 | if run_dir: |
| 184 | os.makedirs(run_dir, exist_ok=True) |
| 185 | # windows cannot take Path objects, only strings |
| 186 | cmd_str_list = [str(c) for c in cmd] |
| 187 | |
| 188 | # Set PYTHONSAFEPATH to prevent adding CWD to sys.path in subprocess Python commands |
| 189 | # This prevents local files from shadowing standard library modules (security issue #1575) |
| 190 | # Supported in Python 3.11+; for earlier versions, the environment variable is ignored |
| 191 | # but those versions are less commonly used and the risk is lower |
| 192 | if len(cmd_str_list) > 0 and "python" in Path(cmd_str_list[0]).name.lower(): |
| 193 | env.setdefault("PYTHONSAFEPATH", "1") |
| 194 | |
| 195 | completed_process = subprocess.run( |
| 196 | cmd_str_list, |
| 197 | env=env, |
| 198 | stdout=subprocess.PIPE if capture_stdout else None, |
| 199 | stderr=subprocess.PIPE if capture_stderr else None, |
| 200 | encoding="utf-8", |
| 201 | text=True, |
| 202 | check=False, |
| 203 | cwd=run_dir, |
| 204 | ) |
| 205 | |
| 206 | if capture_stdout and log_stdout: |
| 207 | _LOGGER.debug(f"stdout: {completed_process.stdout}".rstrip()) |
| 208 | if capture_stderr and log_stderr: |
| 209 | _LOGGER.debug(f"stderr: {completed_process.stderr}".rstrip()) |
| 210 | _LOGGER.debug(f"returncode: {completed_process.returncode}") |
| 211 | |
| 212 | return completed_process |
| 213 | |
| 214 | |
| 215 | def subprocess_post_check(completed_process: "subprocess.CompletedProcess[str]", raise_error: bool = True) -> None: |