Spaces:

Executor-Tyrant-Framework
/

Working-in-a-Codemine

Sleeping

App Files Files Community

Working-in-a-Codemine / tools /shell_tool.py

Executor-Tyrant-Framework

Claude Opus 4.6 (1M context)

Dial in Codemine chassis for OpenRouter + HF Spaces deployment

adbf39e 7 months ago

Raw History Blame Contribute Delete

2.9 kB

# ---- Changelog ----
# [2026-03-29] Chisel/TQB — Block C: ShellTool
# What: shell_execute extracted from RecursiveContextManager
# Why: PRD Block C — single-responsibility tool classes
# [2026-03-29] Razor/TQB — Block A: Security Hardening
# What: PolicyEngine gating replaces joke blocklist, shell=False, output size cap
# Why: PRD Block A — structural security, not advisory string matching
# How: Command split via shlex, subprocess shell=False, PolicyEngine allowlist gate
# -------------------

import logging
import os
import subprocess
from pathlib import Path

logger = logging.getLogger("tools.shell")

# Maximum output size returned from shell commands (chars)
MAX_OUTPUT_SIZE = 50_000

# Shell timeout — configurable via env
SHELL_TIMEOUT = int(os.getenv("CODEMINE_SHELL_TIMEOUT", "60"))


class ShellTool:
    """Shell command execution scoped to repo_path, gated by PolicyEngine."""

    def __init__(self, repo_path: Path, policy_engine=None):
        self.repo_path = repo_path
        self.policy_engine = policy_engine

    def shell_execute(self, command: str) -> str:
        # PolicyEngine gate — structural security
        if self.policy_engine:
            from policy_engine import check_tool_call
            allowed, reason = check_tool_call("shell_execute", {"command": command}, self.repo_path)
            if not allowed:
                logger.warning("shell_execute denied: %s — %s", command[:100], reason)
                return {"status": "error", "tool": "shell", "error": reason, "type": "PermissionError"}

        try:
            # shell=True — pipes and redirects work. Safety is enforced by
            # PolicyEngine's command allowlist, not by shell=False.
            result = subprocess.run(
                command,
                shell=True,
                cwd=str(self.repo_path),
                capture_output=True,
                text=True,
                timeout=SHELL_TIMEOUT,
            )
            stdout = result.stdout
            stderr = result.stderr

            # Output size cap
            if len(stdout) > MAX_OUTPUT_SIZE:
                stdout = stdout[:MAX_OUTPUT_SIZE] + "\n...[truncated]"
            if len(stderr) > MAX_OUTPUT_SIZE:
                stderr = stderr[:MAX_OUTPUT_SIZE] + "\n...[truncated]"

            return f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}"
        except subprocess.TimeoutExpired:
            logger.warning("shell_execute timeout: %s", command[:100])
            return {"status": "error", "tool": "shell", "error": f"Command timed out ({SHELL_TIMEOUT}s limit)", "type": "TimeoutError"}
        except FileNotFoundError as e:
            return {"status": "error", "tool": "shell", "error": f"Command not found: {e}", "type": "FileNotFoundError"}
        except OSError as e:
            return {"status": "error", "tool": "shell", "error": str(e), "type": type(e).__name__}

	# ---- Changelog ----
	# [2026-03-29] Chisel/TQB — Block C: ShellTool
	# What: shell_execute extracted from RecursiveContextManager
	# Why: PRD Block C — single-responsibility tool classes
	# [2026-03-29] Razor/TQB — Block A: Security Hardening
	# What: PolicyEngine gating replaces joke blocklist, shell=False, output size cap
	# Why: PRD Block A — structural security, not advisory string matching
	# How: Command split via shlex, subprocess shell=False, PolicyEngine allowlist gate
	# -------------------

	import logging
	import os
	import subprocess
	from pathlib import Path

	logger = logging.getLogger("tools.shell")

	# Maximum output size returned from shell commands (chars)
	MAX_OUTPUT_SIZE = 50_000

	# Shell timeout — configurable via env
	SHELL_TIMEOUT = int(os.getenv("CODEMINE_SHELL_TIMEOUT", "60"))


	class ShellTool:
	"""Shell command execution scoped to repo_path, gated by PolicyEngine."""

	def __init__(self, repo_path: Path, policy_engine=None):
	self.repo_path = repo_path
	self.policy_engine = policy_engine

	def shell_execute(self, command: str) -> str:
	# PolicyEngine gate — structural security
	if self.policy_engine:
	from policy_engine import check_tool_call
	allowed, reason = check_tool_call("shell_execute", {"command": command}, self.repo_path)
	if not allowed:
	logger.warning("shell_execute denied: %s — %s", command[:100], reason)
	return {"status": "error", "tool": "shell", "error": reason, "type": "PermissionError"}

	try:
	# shell=True — pipes and redirects work. Safety is enforced by
	# PolicyEngine's command allowlist, not by shell=False.
	result = subprocess.run(
	command,
	shell=True,
	cwd=str(self.repo_path),
	capture_output=True,
	text=True,
	timeout=SHELL_TIMEOUT,
	)
	stdout = result.stdout
	stderr = result.stderr

	# Output size cap
	if len(stdout) > MAX_OUTPUT_SIZE:
	stdout = stdout[:MAX_OUTPUT_SIZE] + "\n...[truncated]"
	if len(stderr) > MAX_OUTPUT_SIZE:
	stderr = stderr[:MAX_OUTPUT_SIZE] + "\n...[truncated]"

	return f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}"
	except subprocess.TimeoutExpired:
	logger.warning("shell_execute timeout: %s", command[:100])
	return {"status": "error", "tool": "shell", "error": f"Command timed out ({SHELL_TIMEOUT}s limit)", "type": "TimeoutError"}
	except FileNotFoundError as e:
	return {"status": "error", "tool": "shell", "error": f"Command not found: {e}", "type": "FileNotFoundError"}
	except OSError as e:
	return {"status": "error", "tool": "shell", "error": str(e), "type": type(e).__name__}