import os import inspect import signal import subprocess import time from dataclasses import dataclass from pathlib import Path from typing import Callable from uuid import uuid4 #: Proxy configuration removed from the child environment when network access is denied. PROXY_VARIABLES = ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "FTP_PROXY") #: Any variable ending in one of these never reaches a child process or a log. CREDENTIAL_SUFFIXES = ("API_KEY", "TOKEN", "SECRET", "PASSWORD", "PASSWD", "CREDENTIALS") #: The exception: the providers we drive authenticate through these. Stripping them would #: leave every call failing with 401 and no hint as to why. They are passed through and #: never written to a log or a prompt, which is what the design's rule is actually about. PROVIDER_CREDENTIALS = ( # Direct accounts "OPENAI_API_KEY", "OPENAI_BASE_URL", "CODEX_ACCESS_TOKEN", "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL", "CLAUDE_CODE_OAUTH_TOKEN", # Enterprise backends. Claude can run on Bedrock, Vertex or Foundry, each with its # own credentials; dropping a session token or a service-account path here breaks # every call in a way that looks exactly like "not logged in". "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX", "CLAUDE_CODE_USE_FOUNDRY", "CLAUDE_CODE_SKIP_BEDROCK_AUTH", "CLAUDE_CODE_SKIP_VERTEX_AUTH", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "AWS_PROFILE", "AWS_REGION", "AWS_DEFAULT_REGION", "AWS_BEARER_TOKEN_BEDROCK", "GOOGLE_APPLICATION_CREDENTIALS", "CLOUD_ML_REGION", "ANTHROPIC_VERTEX_PROJECT_ID", "AZURE_OPENAI_API_KEY", "AZURE_OPENAI_ENDPOINT", "AZURE_TENANT_ID", "AZURE_CLIENT_ID", "AZURE_CLIENT_SECRET", ) def collect_secrets(extra_keys: tuple[str, ...] = ()) -> frozenset[str]: """Return non-empty credential values that must be redacted from provider logs.""" keys = list(PROVIDER_CREDENTIALS) + list(extra_keys) return frozenset(value for key in keys if (value := os.environ.get(key, ""))) def run_with_secrets(runner, *args, secrets: frozenset[str], **kwargs): """Call runners while keeping compatibility with narrow test/double signatures.""" parameters = inspect.signature(runner.run).parameters.values() if "secrets" in {parameter.name for parameter in parameters} or any( parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters ): kwargs["secrets"] = secrets return runner.run(*args, **kwargs) def terminate_group(pid: int) -> bool: """Kill a process group by pid. True when a live group was signalled.""" if os.name == "nt": result = subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, check=False) return result.returncode == 0 try: os.killpg(os.getpgid(pid), signal.SIGTERM) return True except (ProcessLookupError, PermissionError, OSError): return False def is_running(pid: int) -> bool: """Whether the recorded run is still alive, used to detect an interrupted task.""" if os.name == "nt": result = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}", "/NH"], capture_output=True, text=True, check=False ) return str(pid) in (result.stdout or "") try: os.kill(pid, 0) return True except ProcessLookupError: return False except PermissionError: return True @dataclass(frozen=True) class ProcessResult: status: str exit_code: int | None stdout_path: Path stderr_path: Path duration_ms: int def child_environment( allow_network: bool, source: dict[str, str] | None = None, extra_allowed: tuple[str, ...] = (), ) -> dict[str, str]: """Build the child environment under the command policy. ponytail: this is a process-level boundary, not a sandbox. It stops accidental credential and proxy inheritance, not a determined process. Upgrade path is a container or an OS sandbox, which the design defers past MVP. """ environment = dict(os.environ if source is None else source) allowed = set(PROVIDER_CREDENTIALS) | {name.upper() for name in extra_allowed} for name in list(environment): upper = name.upper() if upper in allowed: continue if any(upper.endswith(suffix) for suffix in CREDENTIAL_SUFFIXES): del environment[name] elif not allow_network and upper in PROXY_VARIABLES: del environment[name] return environment class ProcessRunner: def __init__( self, log_root: str | Path, allow_network: bool = True, extra_env: tuple[str, ...] = (), ) -> None: self.log_root = Path(log_root) self.allow_network = allow_network #: Extra variable names an enterprise backend needs passed to the CLI. self.extra_env = tuple(extra_env) def run( self, argv: list[str], cwd: str | Path, timeout_s: float, name: str | None = None, on_start: Callable[[int], None] | None = None, stdout_suffix: str = "log", secrets: frozenset[str] = frozenset(), ) -> ProcessResult: self.log_root.mkdir(parents=True, exist_ok=True) key = name or uuid4().hex stdout_path = self.log_root / f"{key}.stdout.{stdout_suffix}" stderr_path = self.log_root / f"{key}.stderr.log" started = time.monotonic() creationflags = subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0 environment = child_environment(self.allow_network, extra_allowed=self.extra_env) process = subprocess.Popen( argv, cwd=str(cwd), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=environment, creationflags=creationflags, start_new_session=os.name != "nt", ) if on_start: on_start(process.pid) try: stdout_bytes, stderr_bytes = process.communicate(timeout=timeout_s) code = process.returncode status = "SUCCESS" if code == 0 else "FAILURE" except subprocess.TimeoutExpired: self._terminate_group(process) try: stdout_bytes, stderr_bytes = process.communicate(timeout=5) except subprocess.TimeoutExpired: if os.name != "nt": try: os.killpg(os.getpgid(process.pid), signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): pass process.kill() try: stdout_bytes, stderr_bytes = process.communicate(timeout=1) except subprocess.TimeoutExpired as drained: # A descendant may still hold the inherited descriptors. The # process itself is dead; close our ends and retain buffered output. stdout_bytes = drained.output or b"" stderr_bytes = drained.stderr or b"" if process.stdout: process.stdout.close() if process.stderr: process.stderr.close() process.wait() code, status = None, "TIMEOUT" stdout_text = stdout_bytes.decode("utf-8", errors="replace") stderr_text = stderr_bytes.decode("utf-8", errors="replace") for secret in sorted((value for value in secrets if value.strip()), key=len, reverse=True): stdout_text = stdout_text.replace(secret, "***") stderr_text = stderr_text.replace(secret, "***") stdout_path.write_text(stdout_text, encoding="utf-8") stderr_path.write_text(stderr_text, encoding="utf-8") return ProcessResult(status, code, stdout_path, stderr_path, int((time.monotonic() - started) * 1000)) @staticmethod def _terminate_group(process: subprocess.Popen) -> None: """Kill the whole group so a timed-out agent leaves no child still editing files.""" if os.name == "nt": subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"], capture_output=True, check=False) else: try: os.killpg(os.getpgid(process.pid), signal.SIGTERM) except (ProcessLookupError, PermissionError): process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill()