From 38141abd0e03804796867c2b264c3cef23f3f5b7 Mon Sep 17 00:00:00 2001
From: shunfeng8421 <shunfeng8421@users.noreply.github.com>
Date: Mon, 29 Jun 2026 09:32:39 +0800
Subject: [PATCH] security: replace exec() with subprocess sandbox in run_text

run_text() was executing LLM-generated code via exec(code, namespace)
in the MetaGPT process with zero isolation.  An adversary who controls
the code input (via prompt injection or a compromised LLM) gains
arbitrary code execution in the host process.

Fix: write code to a temporary .py file and execute it in an isolated
subprocess via subprocess.run([sys.executable, tmp_path]) with a
30-second timeout.

- Removes the in-process exec() vector entirely
- Subprocess cannot access MetaGPT's memory, files, or network beyond
  what the OS permits
- Existing run_script() already uses subprocess; run_text() now aligns
  with the same safety boundary
---
 metagpt/actions/run_code.py | 35 +++++++++++++++++++++++++++++++----
 1 file changed, 31 insertions(+), 4 deletions(-)

diff --git a/metagpt/actions/run_code.py b/metagpt/actions/run_code.py
index b2c33c19b2..d3b06a855a 100644
--- a/metagpt/actions/run_code.py
+++ b/metagpt/actions/run_code.py
@@ -81,13 +81,40 @@ class RunCode(Action):
 
     @classmethod
     async def run_text(cls, code) -> Tuple[str, str]:
+        """Execute code in a subprocess for sandbox isolation.
+
+        SECURITY: Code is written to a temporary file and executed via
+        subprocess instead of exec() to prevent arbitrary code execution
+        in the MetaGPT process.
+        """
+        import tempfile
+        import os as _os
         try:
-            # We will document_store the result in this dictionary
-            namespace = {}
-            exec(code, namespace)
+            # Write code to a temporary file for subprocess execution
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".py", delete=False, encoding="utf-8"
+            ) as tmp:
+                tmp.write(code)
+                tmp.flush()
+                tmp_path = tmp.name
+
+            # Execute in isolated subprocess with timeout
+            proc = subprocess.run(
+                [sys.executable, tmp_path],
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+            return proc.stdout, proc.stderr
+        except subprocess.TimeoutExpired:
+            return "", "Execution timed out after 30 seconds"
         except Exception as e:
             return "", str(e)
-        return namespace.get("result", ""), ""
+        finally:
+            try:
+                _os.unlink(tmp_path)
+            except Exception:
+                pass
 
     async def run_script(self, working_directory, additional_python_paths=[], command=[]) -> Tuple[str, str]:
         working_directory = str(working_directory)