Coverage for src/taipanstack/utils/subprocess.py: 100%
96 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Safe subprocess execution with security guards.
4Provides secure wrappers around subprocess execution with
5command validation, timeout handling, and retry logic.
6"""
8import math
9import os
10import shutil
11import subprocess # nosec B404
12import time
13from collections.abc import Sequence
14from dataclasses import dataclass
15from pathlib import Path
17from taipanstack.security.guards import (
18 SecurityError,
19 guard_command_injection,
20)
23@dataclass(frozen=True)
24class SafeCommandResult:
25 """Result of a safe command execution.
27 Attributes:
28 command: The executed command.
29 returncode: Exit code of the command.
30 stdout: Standard output.
31 stderr: Standard error.
32 success: Whether the command succeeded (returncode == 0).
33 duration_seconds: How long the command took.
35 """
37 command: list[str]
38 returncode: int
39 stdout: str = ""
40 stderr: str = ""
41 duration_seconds: float = 0.0
43 @property
44 def success(self) -> bool:
45 """Check if command succeeded."""
46 return self.returncode == 0
48 def raise_on_error(self) -> "SafeCommandResult":
49 """Raise an exception if command failed.
51 Returns:
52 Self if successful.
54 Raises:
55 subprocess.CalledProcessError: If command failed.
57 """
58 if not self.success:
59 raise subprocess.CalledProcessError(
60 self.returncode,
61 self.command,
62 self.stdout,
63 self.stderr,
64 )
65 return self
68# Default allowed commands whitelist
69DEFAULT_ALLOWED_COMMANDS: frozenset[str] = frozenset(
70 {
71 # Python/Poetry
72 "python",
73 "python3",
74 "pip",
75 "pip3",
76 "poetry",
77 "pipx",
78 # Version control
79 "git",
80 # Build tools
81 "make",
82 # Testing
83 "pytest",
84 "mypy",
85 "ruff",
86 "bandit",
87 "safety",
88 "semgrep",
89 "pre-commit",
90 # System
91 "echo",
92 "cat",
93 "ls",
94 "pwd",
95 "mkdir",
96 "rm",
97 "cp",
98 "mv",
99 "touch",
100 "chmod",
101 "which",
102 },
103)
106def _validate_and_resolve_command(
107 command: Sequence[str],
108 allowed_commands: Sequence[str] | None,
109) -> list[str]:
110 """Validate and resolve a command.
112 Args:
113 command: Command and arguments as a sequence.
114 allowed_commands: Whitelist of allowed commands.
116 Returns:
117 The validated command as a list.
119 Raises:
120 SecurityError: If validation fails.
122 """
123 cmd_list = list(command)
125 if not cmd_list:
126 raise SecurityError(
127 "Empty command is not allowed",
128 guard_name="safe_command",
129 )
131 if allowed_commands is not None:
132 whitelist = list(allowed_commands)
133 else:
134 whitelist = list(DEFAULT_ALLOWED_COMMANDS)
136 validated_cmd = guard_command_injection(cmd_list, allowed_commands=whitelist)
138 base_command = validated_cmd[0]
139 if not shutil.which(base_command):
140 raise SecurityError(
141 f"Command not found: {base_command}",
142 guard_name="safe_command",
143 value=base_command,
144 )
146 return validated_cmd
149def _get_allowed_keys(allowed_env_vars: Sequence[str] | None) -> set[str]:
150 """Get the allowed environment variable keys.
152 Args:
153 allowed_env_vars: Whitelist of allowed environment variable names.
155 Returns:
156 Set of allowed keys in uppercase.
158 """
159 if allowed_env_vars is None:
160 return {"PATH"}
161 return {k.upper() for k in allowed_env_vars}
164def _get_env_to_filter(env: dict[str, str] | None) -> dict[str, str]:
165 """Get the environment to filter."""
166 return env if env is not None else dict(os.environ)
169def _build_filtered_env(
170 env_to_filter: dict[str, str], allowed_keys: set[str]
171) -> dict[str, str]:
172 """Build the filtered environment dictionary."""
173 if not allowed_keys:
174 return {}
175 return {k: str(v) for k, v in env_to_filter.items() if k.upper() in allowed_keys}
178def _filter_environment(
179 env: dict[str, str] | None,
180 allowed_env_vars: Sequence[str] | None = None,
181) -> dict[str, str]:
182 """Filter environment variables for safe execution using a whitelist approach.
184 Args:
185 env: Environment variables to filter. If None, uses os.environ.
186 allowed_env_vars: Whitelist of allowed environment variable names.
187 If None, only minimal necessary variables (e.g. PATH) are inherited.
188 Pass an empty list for a completely empty environment.
190 Returns:
191 Filtered environment variables containing only allowed keys.
193 """
194 env_to_filter = _get_env_to_filter(env)
195 allowed_keys = _get_allowed_keys(allowed_env_vars)
196 return _build_filtered_env(env_to_filter, allowed_keys)
199def _extract_timeout_stdout(e: subprocess.TimeoutExpired) -> str:
200 """Extract stdout from a TimeoutExpired exception safely.
202 Args:
203 e: The TimeoutExpired exception.
205 Returns:
206 The extracted stdout string.
208 """
209 if not hasattr(e, "stdout") or e.stdout is None:
210 return ""
211 if isinstance(e.stdout, str): # type: ignore[unreachable]
212 return e.stdout # type: ignore[unreachable]
213 return e.stdout.decode("utf-8", errors="replace")
216def _execute_command(
217 validated_cmd: list[str],
218 cwd: Path | None,
219 timeout: float,
220 capture_output: bool,
221 safe_env: dict[str, str],
222) -> SafeCommandResult:
223 """Execute the command and handle TimeoutExpired.
225 Args:
226 validated_cmd: Validated command list.
227 cwd: Resolved working directory.
228 timeout: Execution timeout.
229 capture_output: Capture stdout/stderr.
230 safe_env: Filtered environment variables.
232 Returns:
233 The execution result.
235 """
236 start_time = time.time()
238 try:
239 result = subprocess.run( # nosec B603
240 validated_cmd,
241 cwd=cwd,
242 timeout=timeout,
243 capture_output=capture_output,
244 text=True,
245 encoding="utf-8",
246 env=safe_env,
247 check=False,
248 )
249 except subprocess.TimeoutExpired as e:
250 duration = time.time() - start_time
251 return SafeCommandResult(
252 command=validated_cmd,
253 returncode=-1,
254 stdout=_extract_timeout_stdout(e),
255 stderr=f"Command timed out after {timeout}s",
256 duration_seconds=duration,
257 )
259 duration = time.time() - start_time
261 return SafeCommandResult(
262 command=validated_cmd,
263 returncode=result.returncode,
264 stdout=result.stdout or "",
265 stderr=result.stderr or "",
266 duration_seconds=duration,
267 )
270def _validate_cwd(cwd: Path | str | None) -> Path | None:
271 """Resolve and validate the working directory."""
272 if cwd is None:
273 return None
274 resolved_cwd = Path(cwd).resolve()
275 if not resolved_cwd.exists():
276 raise SecurityError(
277 f"Working directory does not exist: {cwd}",
278 guard_name="safe_command",
279 )
280 return resolved_cwd
283def _handle_dry_run(validated_cmd: list[str]) -> SafeCommandResult:
284 """Return a SafeCommandResult for a dry run execution."""
285 return SafeCommandResult(
286 command=validated_cmd,
287 returncode=0,
288 stdout=f"[DRY-RUN] Would execute: {' '.join(validated_cmd)}",
289 stderr="",
290 duration_seconds=0.0,
291 )
294def _check_timeout_type(timeout: float) -> None:
295 """Check if timeout is a number."""
296 if not isinstance(timeout, (int, float)):
297 raise TypeError("timeout must be a finite non-negative number")
300def _check_timeout_value(timeout: float) -> None:
301 """Check if timeout is finite and non-negative."""
302 if not math.isfinite(timeout) or timeout < 0:
303 raise ValueError("timeout must be a finite non-negative number")
306def _validate_timeout(timeout: float) -> None:
307 """Validate that the timeout is a finite, non-negative number.
309 Args:
310 timeout: The timeout value to validate.
312 Raises:
313 ValueError: If timeout is invalid.
315 """
316 if timeout is None:
317 return # type: ignore[unreachable]
319 _check_timeout_type(timeout)
320 _check_timeout_value(timeout)
323def run_safe_command(
324 command: Sequence[str],
325 *,
326 cwd: Path | str | None = None,
327 timeout: float = 300.0,
328 capture_output: bool = True,
329 check: bool = False,
330 allowed_commands: Sequence[str] | None = None,
331 env: dict[str, str] | None = None,
332 allowed_env_vars: Sequence[str] | None = None,
333 dry_run: bool = False,
334) -> SafeCommandResult:
335 """Execute a command safely with security guards.
337 This function provides a secure wrapper around subprocess.run
338 with command injection protection, timeout handling, and
339 optional command whitelisting.
341 Args:
342 command: Command and arguments as a sequence.
343 cwd: Working directory for the command.
344 timeout: Maximum execution time in seconds.
345 capture_output: Whether to capture stdout/stderr.
346 check: Whether to raise on non-zero exit.
347 allowed_commands: Whitelist of allowed commands.
348 env: Environment variables to set.
349 allowed_env_vars: Whitelist of allowed environment variables.
350 dry_run: If True, don't actually execute the command.
352 Returns:
353 SafeCommandResult with execution details.
355 Raises:
356 SecurityError: If command validation fails.
357 subprocess.TimeoutExpired: If command times out.
358 subprocess.CalledProcessError: If check=True and command fails.
360 Example:
361 >>> result = run_safe_command(["poetry", "install"])
362 >>> if result.success:
363 ... print("Installation complete!")
365 """
366 # Security Enhancement: Prevent DoS via infinite blocking behaviors or
367 # unhandled exceptions from threading/asyncio primitives by explicitly
368 # validating that timeout is a finite, non-negative number.
369 _validate_timeout(timeout)
371 validated_cmd = _validate_and_resolve_command(command, allowed_commands)
372 safe_env = _filter_environment(env, allowed_env_vars)
374 if dry_run:
375 return _handle_dry_run(validated_cmd)
377 resolved_cwd = _validate_cwd(cwd)
379 safe_result = _execute_command(
380 validated_cmd,
381 resolved_cwd,
382 timeout,
383 capture_output,
384 safe_env,
385 )
387 if check:
388 safe_result.raise_on_error()
390 return safe_result