diff options
Diffstat (limited to 'autogpt/commands/execute_code.py')
| -rw-r--r-- | autogpt/commands/execute_code.py | 303 |
1 files changed, 0 insertions, 303 deletions
diff --git a/autogpt/commands/execute_code.py b/autogpt/commands/execute_code.py deleted file mode 100644 index aad93193e..000000000 --- a/autogpt/commands/execute_code.py +++ /dev/null @@ -1,303 +0,0 @@ -"""Execute code in a Docker container""" -import os -import subprocess -from pathlib import Path - -import docker -from docker.errors import DockerException, ImageNotFound -from docker.models.containers import Container as DockerContainer - -from autogpt.agent.agent import Agent -from autogpt.command_decorator import command -from autogpt.config import Config -from autogpt.logs import logger - -from .decorators import sanitize_path_arg - -ALLOWLIST_CONTROL = "allowlist" -DENYLIST_CONTROL = "denylist" - - -@command( - "execute_python_code", - "Creates a Python file and executes it", - { - "code": { - "type": "string", - "description": "The Python code to run", - "required": True, - }, - "name": { - "type": "string", - "description": "A name to be given to the python file", - "required": True, - }, - }, -) -def execute_python_code(code: str, name: str, agent: Agent) -> str: - """Create and execute a Python file in a Docker container and return the STDOUT of the - executed code. If there is any data that needs to be captured use a print statement - - Args: - code (str): The Python code to run - name (str): A name to be given to the Python file - - Returns: - str: The STDOUT captured from the code when it ran - """ - ai_name = agent.ai_config.ai_name - code_dir = agent.workspace.get_path(Path(ai_name, "executed_code")) - os.makedirs(code_dir, exist_ok=True) - - if not name.endswith(".py"): - name = name + ".py" - - # The `name` arg is not covered by @sanitize_path_arg, - # so sanitization must be done here to prevent path traversal. - file_path = agent.workspace.get_path(code_dir / name) - if not file_path.is_relative_to(code_dir): - return "Error: 'name' argument resulted in path traversal, operation aborted" - - try: - with open(file_path, "w+", encoding="utf-8") as f: - f.write(code) - - return execute_python_file(str(file_path), agent) - except Exception as e: - return f"Error: {str(e)}" - - -@command( - "execute_python_file", - "Executes an existing Python file", - { - "filename": { - "type": "string", - "description": "The name of te file to execute", - "required": True, - }, - }, -) -@sanitize_path_arg("filename") -def execute_python_file(filename: str, agent: Agent) -> str: - """Execute a Python file in a Docker container and return the output - - Args: - filename (str): The name of the file to execute - - Returns: - str: The output of the file - """ - logger.info( - f"Executing python file '{filename}' in working directory '{agent.config.workspace_path}'" - ) - - if not filename.endswith(".py"): - return "Error: Invalid file type. Only .py files are allowed." - - file_path = Path(filename) - if not file_path.is_file(): - # Mimic the response that you get from the command line so that it's easier to identify - return ( - f"python: can't open file '{filename}': [Errno 2] No such file or directory" - ) - - if we_are_running_in_a_docker_container(): - logger.debug( - f"Auto-GPT is running in a Docker container; executing {file_path} directly..." - ) - result = subprocess.run( - ["python", str(file_path)], - capture_output=True, - encoding="utf8", - cwd=agent.config.workspace_path, - ) - if result.returncode == 0: - return result.stdout - else: - return f"Error: {result.stderr}" - - logger.debug("Auto-GPT is not running in a Docker container") - try: - client = docker.from_env() - # You can replace this with the desired Python image/version - # You can find available Python images on Docker Hub: - # https://hub.docker.com/_/python - image_name = "python:3-alpine" - try: - client.images.get(image_name) - logger.debug(f"Image '{image_name}' found locally") - except ImageNotFound: - logger.info( - f"Image '{image_name}' not found locally, pulling from Docker Hub..." - ) - # Use the low-level API to stream the pull response - low_level_client = docker.APIClient() - for line in low_level_client.pull(image_name, stream=True, decode=True): - # Print the status and progress, if available - status = line.get("status") - progress = line.get("progress") - if status and progress: - logger.info(f"{status}: {progress}") - elif status: - logger.info(status) - - logger.debug(f"Running {file_path} in a {image_name} container...") - container: DockerContainer = client.containers.run( - image_name, - ["python", str(file_path.relative_to(agent.workspace.root))], - volumes={ - agent.config.workspace_path: { - "bind": "/workspace", - "mode": "ro", - } - }, - working_dir="/workspace", - stderr=True, - stdout=True, - detach=True, - ) # type: ignore - - container.wait() - logs = container.logs().decode("utf-8") - container.remove() - - # print(f"Execution complete. Output: {output}") - # print(f"Logs: {logs}") - - return logs - - except DockerException as e: - logger.warn( - "Could not run the script in a container. If you haven't already, please install Docker https://docs.docker.com/get-docker/" - ) - return f"Error: {str(e)}" - - except Exception as e: - return f"Error: {str(e)}" - - -def validate_command(command: str, config: Config) -> bool: - """Validate a command to ensure it is allowed - - Args: - command (str): The command to validate - config (Config): The config to use to validate the command - - Returns: - bool: True if the command is allowed, False otherwise - """ - if not command: - return False - - command_name = command.split()[0] - - if config.shell_command_control == ALLOWLIST_CONTROL: - return command_name in config.shell_allowlist - else: - return command_name not in config.shell_denylist - - -@command( - "execute_shell", - "Executes a Shell Command, non-interactive commands only", - { - "command_line": { - "type": "string", - "description": "The command line to execute", - "required": True, - } - }, - enabled=lambda config: config.execute_local_commands, - disabled_reason="You are not allowed to run local shell commands. To execute" - " shell commands, EXECUTE_LOCAL_COMMANDS must be set to 'True' " - "in your config file: .env - do not attempt to bypass the restriction.", -) -def execute_shell(command_line: str, agent: Agent) -> str: - """Execute a shell command and return the output - - Args: - command_line (str): The command line to execute - - Returns: - str: The output of the command - """ - if not validate_command(command_line, agent.config): - logger.info(f"Command '{command_line}' not allowed") - return "Error: This Shell Command is not allowed." - - current_dir = Path.cwd() - # Change dir into workspace if necessary - if not current_dir.is_relative_to(agent.config.workspace_path): - os.chdir(agent.config.workspace_path) - - logger.info( - f"Executing command '{command_line}' in working directory '{os.getcwd()}'" - ) - - result = subprocess.run(command_line, capture_output=True, shell=True) - output = f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" - - # Change back to whatever the prior working dir was - - os.chdir(current_dir) - return output - - -@command( - "execute_shell_popen", - "Executes a Shell Command, non-interactive commands only", - { - "query": { - "type": "string", - "description": "The search query", - "required": True, - } - }, - lambda config: config.execute_local_commands, - "You are not allowed to run local shell commands. To execute" - " shell commands, EXECUTE_LOCAL_COMMANDS must be set to 'True' " - "in your config. Do not attempt to bypass the restriction.", -) -def execute_shell_popen(command_line, agent: Agent) -> str: - """Execute a shell command with Popen and returns an english description - of the event and the process id - - Args: - command_line (str): The command line to execute - - Returns: - str: Description of the fact that the process started and its id - """ - if not validate_command(command_line, agent.config): - logger.info(f"Command '{command_line}' not allowed") - return "Error: This Shell Command is not allowed." - - current_dir = os.getcwd() - # Change dir into workspace if necessary - if agent.config.workspace_path not in current_dir: - os.chdir(agent.config.workspace_path) - - logger.info( - f"Executing command '{command_line}' in working directory '{os.getcwd()}'" - ) - - do_not_show_output = subprocess.DEVNULL - process = subprocess.Popen( - command_line, shell=True, stdout=do_not_show_output, stderr=do_not_show_output - ) - - # Change back to whatever the prior working dir was - - os.chdir(current_dir) - - return f"Subprocess started with PID:'{str(process.pid)}'" - - -def we_are_running_in_a_docker_container() -> bool: - """Check if we are running in a Docker container - - Returns: - bool: True if we are running in a Docker container, False otherwise - """ - return os.path.exists("/.dockerenv") |