Spaces:
Running
Running
File size: 3,796 Bytes
81aa0b5 | 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 | """Bash subprocess tool with timeout and output capture."""
from __future__ import annotations
import os
import shlex
import subprocess
from typing import Any
from code.tools.fs import _resolve_safe, get_workspace_root
# ─── Safety: commands that are forbidden by default ────────────────────
_BLOCKED_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"rm -rf $HOME",
":(){:|:&};:",
"mkfs",
"dd if=/dev/zero of=/dev/",
"> /dev/sda",
"shutdown",
"reboot",
"halt",
]
# Default env vars to scrub for safety
_ENV_SCRUB = {"HF_TOKEN", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "AWS_SECRET_ACCESS_KEY"}
def _is_safe_command(cmd: str) -> tuple[bool, str]:
"""Check if a command is safe to run."""
stripped = cmd.strip()
if not stripped:
return False, "Empty command"
for pat in _BLOCKED_PATTERNS:
if pat in stripped:
return False, f"Blocked pattern detected: {pat}"
return True, ""
def run_bash(
command: str,
cwd: str | None = None,
timeout: int = 30,
env_extra: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Run a shell command in the workspace.
Args:
command: Shell command to execute.
cwd: Working directory (relative to workspace root, defaults to workspace).
timeout: Max seconds before killing the process.
env_extra: Extra environment variables.
Returns:
dict with: stdout, stderr, returncode, timed_out
"""
try:
safe, reason = _is_safe_command(command)
if not safe:
return {
"success": False,
"stdout": "",
"stderr": reason,
"returncode": -1,
"timed_out": False,
}
if cwd:
work_dir = _resolve_safe(cwd)
else:
work_dir = get_workspace_root()
# Build env: scrub secrets, add extras
env = {k: v for k, v in os.environ.items() if k not in _ENV_SCRUB}
if env_extra:
env.update(env_extra)
# Run with bash -c for full shell semantics
completed = subprocess.run(
["bash", "-c", command],
cwd=work_dir,
env=env,
capture_output=True,
text=True,
timeout=timeout,
check=False,
)
# Truncate huge outputs
stdout = completed.stdout
stderr = completed.stderr
if len(stdout) > 50_000:
stdout = stdout[:50_000] + f"\n... truncated ({len(stdout) - 50_000} chars) ..."
if len(stderr) > 50_000:
stderr = stderr[:50_000] + f"\n... truncated ({len(stderr) - 50_000} chars) ..."
return {
"success": completed.returncode == 0,
"stdout": stdout,
"stderr": stderr,
"returncode": completed.returncode,
"timed_out": False,
"command": command,
"cwd": cwd or ".",
}
except subprocess.TimeoutExpired as exc:
stdout = exc.stdout or "" if isinstance(exc.stdout, str) else (exc.stdout or b"").decode("utf-8", errors="replace")
stderr = exc.stderr or "" if isinstance(exc.stderr, str) else (exc.stderr or b"").decode("utf-8", errors="replace")
return {
"success": False,
"stdout": stdout,
"stderr": f"Timeout after {timeout}s\n{stderr}",
"returncode": -1,
"timed_out": True,
"command": command,
}
except Exception as exc:
return {
"success": False,
"stdout": "",
"stderr": str(exc),
"returncode": -1,
"timed_out": False,
"command": command,
}
|