tiphys.tools.shell

  1"""
  2Shell command execution tool for agents.
  3
  4Provides safe command execution with policy checks.
  5"""
  6
  7import asyncio
  8from pathlib import Path
  9
 10import structlog
 11
 12from tiphys.config import TiphysConfig, load_config
 13from tiphys.security.analyze import RiskLevel, analyze_command
 14from tiphys.security.policy import ToolContext, ToolPermission, create_default_policy
 15from tiphys.tools.base import ToolDefinition, tool
 16
 17logger = structlog.get_logger(__name__)
 18
 19# Maximum output size (1MB)
 20MAX_OUTPUT_SIZE = 1024 * 1024
 21
 22# Default timeout (60 seconds)
 23DEFAULT_TIMEOUT = 60.0
 24
 25
 26@tool(
 27    name="run_command",
 28    description="Execute a shell command and return its output. Use with caution.",
 29    requires_permission=True,
 30)
 31async def run_command(
 32    command: str,
 33    cwd: str | None = None,
 34    timeout: float = DEFAULT_TIMEOUT,
 35) -> str:
 36    """
 37    Execute a shell command.
 38
 39    Args:
 40        command: The shell command to execute.
 41        cwd: Working directory for the command.
 42        timeout: Timeout in seconds (max 300).
 43
 44    Returns:
 45        Command output (stdout + stderr) or error message.
 46    """
 47    config = load_config()
 48
 49    # Validate timeout
 50    timeout = min(timeout, 300.0)
 51
 52    # Analyze the command for security
 53    analysis = analyze_command(command, config.allowed_commands)
 54
 55    # Log command analysis
 56    await logger.ainfo(
 57        "Command analysis",
 58        command=command[:100],
 59        risk_level=analysis.risk_level.value,
 60        binaries=analysis.binaries,
 61        is_allowed=analysis.is_allowed,
 62    )
 63
 64    # Check exec_mode
 65    if config.exec_mode == "deny":
 66        return "Error: Command execution is disabled (exec_mode=deny)"
 67
 68    # Check allowlist mode
 69    if config.exec_mode == "allowlist" and not analysis.is_allowed:
 70        denied_binaries = [b for b in analysis.binaries if b not in config.allowed_commands]
 71        return f"Error: Command blocked. Binaries not in allowlist: {denied_binaries}"
 72
 73    # Block critical risk commands
 74    if analysis.risk_level == RiskLevel.CRITICAL:
 75        return f"Error: Command blocked due to critical risk. Reasons: {analysis.reasons}"
 76
 77    # Validate working directory
 78    work_dir: Path | None = None
 79    if cwd:
 80        try:
 81            work_dir = Path(cwd).resolve()
 82            if not work_dir.is_dir():
 83                return f"Error: Working directory does not exist: {cwd}"
 84        except Exception as e:
 85            return f"Error: Invalid working directory: {e}"
 86
 87    # Execute the command
 88    try:
 89        proc = await asyncio.create_subprocess_shell(
 90            command,
 91            stdout=asyncio.subprocess.PIPE,
 92            stderr=asyncio.subprocess.PIPE,
 93            cwd=work_dir,
 94        )
 95
 96        try:
 97            stdout, stderr = await asyncio.wait_for(
 98                proc.communicate(),
 99                timeout=timeout,
100            )
101        except TimeoutError:
102            proc.kill()
103            await proc.wait()
104            return f"Error: Command timed out after {timeout} seconds"
105
106        # Decode output
107        stdout_str = stdout.decode("utf-8", errors="replace")
108        stderr_str = stderr.decode("utf-8", errors="replace")
109
110        # Combine output
111        output_parts: list[str] = []
112
113        if stdout_str:
114            output_parts.append(stdout_str)
115
116        if stderr_str:
117            if output_parts:
118                output_parts.append("\n--- stderr ---\n")
119            output_parts.append(stderr_str)
120
121        output = "".join(output_parts)
122
123        # Truncate if too long
124        if len(output) > MAX_OUTPUT_SIZE:
125            output = output[:MAX_OUTPUT_SIZE] + f"\n... (truncated, {len(output)} total bytes)"
126
127        # Add exit code info
128        if proc.returncode != 0:
129            output += f"\n\n[Exit code: {proc.returncode}]"
130
131        return output if output.strip() else "(no output)"
132
133    except PermissionError:
134        return "Error: Permission denied executing command"
135    except FileNotFoundError:
136        return f"Error: Command not found: {command.split()[0] if command else command}"
137    except Exception as e:
138        await logger.aerror("Command execution failed", command=command[:100], error=str(e))
139        return f"Error executing command: {e}"
140
141
142async def check_command_permission(
143    command: str,
144    context: ToolContext,
145    config: TiphysConfig | None = None,
146) -> tuple[bool, str]:
147    """
148    Check if a command is allowed by policy.
149
150    Args:
151        command: The command to check.
152        context: The tool execution context.
153        config: Optional configuration.
154
155    Returns:
156        Tuple of (is_allowed, reason).
157    """
158    if config is None:
159        config = load_config()
160
161    # Create policy engine
162    engine = create_default_policy(config)
163
164    # Check tool permission
165    permission = engine.check("run_command", context)
166
167    if permission == ToolPermission.DENIED:
168        return False, "Command execution denied by policy"
169
170    # Analyze command
171    analysis = analyze_command(command, config.allowed_commands)
172
173    if config.exec_mode == "deny":
174        return False, "Command execution disabled"
175
176    if config.exec_mode == "allowlist" and not analysis.is_allowed:
177        return False, f"Binaries not in allowlist: {analysis.binaries}"
178
179    if analysis.risk_level == RiskLevel.CRITICAL:
180        return False, f"Critical risk: {analysis.reasons}"
181
182    if permission == ToolPermission.ASK:
183        return True, "Requires user approval"
184
185    return True, "Allowed"
186
187
188# Export tool definitions
189def get_shell_tools() -> list[ToolDefinition]:
190    """Get all shell tool definitions."""
191    tools: list[ToolDefinition] = []
192
193    if hasattr(run_command, "_tool_definition"):
194        tools.append(run_command._tool_definition)
195
196    return tools