File size: 6,054 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """Module for subprocess runner."""
from __future__ import annotations
import subprocess
from dataclasses import dataclass
from typing import Any
from apps.core.exceptions import ExternalServiceError
from apps.core.security.scrub import scrub_text
@dataclass(frozen=True)
class SubprocessOutput:
stdout: str
stderr: str
returncode: int
class SubprocessRunner:
def __init__(self, *, allowed_programs: set[str] | None = None, max_output_chars: int = 20_000) -> None:
self._allowed_programs = allowed_programs
self._max_output_chars = int(max_output_chars) if max_output_chars is not None else 20_000
def _truncate(self, value: str) -> str:
if not value:
return ""
if self._max_output_chars <= 0:
return ""
if len(value) <= self._max_output_chars:
return value
return value[: self._max_output_chars] + "...(truncated)"
def _validate_args(self, args: list[str]) -> None:
if not isinstance(args, list) or not args or not all(isinstance(a, str) for a in args):
raise ExternalServiceError(message="外部命令参数无效", code="SUBPROCESS_INVALID_ARGS", errors={})
program = str(args[0] or "").strip()
if not program:
raise ExternalServiceError(message="外部命令参数无效", code="SUBPROCESS_INVALID_ARGS", errors={})
if self._allowed_programs is not None and program not in self._allowed_programs:
raise ExternalServiceError(
message="外部命令不允许",
code="SUBPROCESS_NOT_ALLOWED",
errors={"command": scrub_text(program)},
)
def run(
self,
*,
args: list[str],
timeout_seconds: float | None = None,
check: bool = False,
text: bool = True,
) -> SubprocessOutput:
self._validate_args(args)
try:
cp = subprocess.run( # 安全：args 已通过 _validate_args 白名单校验，shell=False
args,
check=check,
capture_output=True,
text=text,
timeout=timeout_seconds,
)
except subprocess.CalledProcessError as e:
stdout = e.stdout or ""
stderr = e.stderr or ""
if isinstance(stdout, bytes):
stdout = stdout.decode("utf-8", errors="replace")
if isinstance(stderr, bytes):
stderr = stderr.decode("utf-8", errors="replace")
raise ExternalServiceError(
message="外部命令执行失败",
code="SUBPROCESS_NONZERO_EXIT",
errors={
"command": scrub_text(str(args[0] if args else "")),
"returncode": int(getattr(e, "returncode", 0) or 0),
"stdout": self._truncate(scrub_text(str(stdout))),
"stderr": self._truncate(scrub_text(str(stderr))),
},
) from e
except subprocess.TimeoutExpired as e:
raise ExternalServiceError(
message="外部命令执行超时",
code="SUBPROCESS_TIMEOUT",
errors={
"command": scrub_text(str(args[0] if args else "")),
"timeout_seconds": timeout_seconds,
"detail": self._truncate(scrub_text(str(e))),
},
) from e
except FileNotFoundError as e:
raise ExternalServiceError(
message="外部命令不存在",
code="SUBPROCESS_NOT_FOUND",
errors={
"command": scrub_text(str(args[0] if args else "")),
"detail": self._truncate(scrub_text(str(e))),
},
) from e
except Exception as e:
raise ExternalServiceError(
message="外部命令执行失败",
code="SUBPROCESS_FAILED",
errors={
"command": scrub_text(str(args[0] if args else "")),
"detail": self._truncate(scrub_text(str(e))),
},
) from e
stdout = cp.stdout or ""
stderr = cp.stderr or ""
if isinstance(stdout, bytes):
stdout = stdout.decode("utf-8", errors="replace")
if isinstance(stderr, bytes):
stderr = stderr.decode("utf-8", errors="replace")
return SubprocessOutput(
stdout=self._truncate(scrub_text(str(stdout))),
stderr=self._truncate(scrub_text(str(stderr))),
returncode=int(cp.returncode or 0),
)
def popen(self, *, args: list[str], **kwargs: Any) -> subprocess.Popen[Any]:
self._validate_args(args)
if "shell" in kwargs and kwargs.get("shell"):
raise ExternalServiceError(
message="外部命令执行参数不安全",
code="SUBPROCESS_UNSAFE_OPTIONS",
errors={"command": scrub_text(str(args[0] if args else "")), "option": "shell"},
)
try:
return subprocess.Popen(
args, **kwargs
) # 安全：args 已通过 _validate_args 白名单校验，且明确拒绝 shell=True
except FileNotFoundError as e:
raise ExternalServiceError(
message="外部命令不存在",
code="SUBPROCESS_NOT_FOUND",
errors={
"command": scrub_text(str(args[0] if args else "")),
"detail": self._truncate(scrub_text(str(e))),
},
) from e
except Exception as e:
raise ExternalServiceError(
message="外部命令执行失败",
code="SUBPROCESS_FAILED",
errors={
"command": scrub_text(str(args[0] if args else "")),
"detail": self._truncate(scrub_text(str(e))),
},
) from e
|