Bash 工具

import subprocess
import threading
import queue


class BashSession:
    def __init__(self):
        self.process = subprocess.Popen(
            ["/bin/bash"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=0,
        )
        self.output_queue = queue.Queue()
        self.error_queue = queue.Queue()
        self._start_readers()

def execute_command(self, command):
    # Send command to bash
    self.process.stdin.write(command + "\n")
    self.process.stdin.flush()

    # Capture output with timeout
    output = self._read_output(timeout=10)
    return output

for content in response.content:
    if content.type == "tool_use" and content.name == "bash":
        if content.input.get("restart"):
            bash_session.restart()
            result = "Bash session restarted"
        else:
            command = content.input.get("command")
            result = bash_session.execute_command(command)

        # Return result to Claude
        tool_result = {
            "type": "tool_result",
            "tool_use_id": content.id,
            "content": result,
        }

import shlex

ALLOWED_COMMANDS = {"ls", "cat", "echo", "pwd", "grep", "find", "wc", "head", "tail"}
SHELL_OPERATORS = {"&&", "||", "|", ";", "&", ">", "<", ">>"}


def validate_command(command):
    # Allow only commands from an explicit allowlist
    try:
        tokens = shlex.split(command)
    except ValueError:
        return False, "Could not parse command"

    if not tokens:
        return False, "Empty command"

    executable = tokens[0]
    if executable not in ALLOWED_COMMANDS:
        return False, f"Command '{executable}' is not in the allowlist"

    # Reject shell operators that would chain additional commands
    for token in tokens[1:]:
        if token in SHELL_OPERATORS or token.startswith(("$", "`")):
            return False, f"Shell operator '{token}' is not allowed"

    return True, None