Using AutoGen's LocalCommandLineCodeExecutor safely

AutoGen’s LocalCommandLineCodeExecutor lets agents write and run code on the host machine, which is powerful and dangerous. This autogen localcommandlinecodeexecutor tutorial walks through hardening it for production use: sandboxing the filesystem, restricting commands, capping resources, and adding observability so you can ship code-executing agents without handing attackers a shell.

Step 1: understand the threat model

LocalCommandLineCodeExecutor executes arbitrary Python snippets in a subprocess. By default it inherits the parent process’s environment, filesystem access, and network reach. A compromised or hallucinating agent can delete files, exfiltrate secrets, install packages, or pivot to internal services. Treat every execution as untrusted input.

The executor does not sandbox by default. You must layer defenses: filesystem isolation, command allowlists, resource quotas, and audit logging. Defense in depth means any single bypass still leaves the attacker contained.

Step 2: create a dedicated execution user and directory

Run the executor under a low-privilege OS user with a stripped-down home directory. On Linux:

sudo useradd --system --no-create-home --shell /usr/sbin/nologin autogen-exec
sudo mkdir -p /var/lib/autogen/sandbox
sudo chown autogen-exec:autogen-exec /var/lib/autogen/sandbox
sudo chmod 700 /var/lib/autogen/sandbox

The sandbox directory is the only path the executor should ever touch. Mount it noexec,nosuid,nodev if your kernel supports it:

# /etc/fstab entry (adjust device)
/dev/nvme0n1p2  /var/lib/autogen/sandbox  ext4  defaults,noexec,nosuid,nodev  0  2

Verify the mount options:

findmnt -o TARGET,OPTIONS /var/lib/autogen/sandbox

You should see noexec,nosuid,nodev in the output. This prevents binary execution from the sandbox even if an attacker writes an ELF file.

Step 3: wrap the executor in a container or VM

OS-level isolation beats user-level isolation. Run the executor inside a container with a read-only root filesystem, dropped capabilities, and no network. Example Dockerfile:

FROM python:3.11-slim

# Install only the runtime deps your agents need
RUN pip install --no-cache-dir autogen-agentchat==0.2.0

# Create the sandbox user inside the container
RUN useradd --system --no-create-home --shell /usr/sbin/nologin autogen-exec \
    && mkdir -p /sandbox \
    && chown autogen-exec:autogen-exec /sandbox

USER autogen-exec
WORKDIR /sandbox
ENTRYPOINT ["python", "-m", "autogen_core.code_executor"]

Build and run with strict constraints:

docker build -t autogen-executor:latest .

docker run -d \
  --name autogen-executor \
  --user autogen-exec \
  --read-only \
  --tmpfs /sandbox:noexec,nosuid,nodev,size=100m \
  --cpus="0.5" \
  --memory="256m" \
  --pids-limit=50 \
  --network=none \
  --cap-drop=ALL \
  --security-opt=no-new-privileges \
  autogen-executor:latest

Key flags:

• --read-only — rootfs immutable
• --tmpfs — sandbox lives in memory, disappears on stop
• --cpus / --memory — hard resource ceilings
• --pids-limit — prevents fork bombs
• --network=none — no outbound or inbound traffic
• --cap-drop=ALL — no Linux capabilities
• --security-opt=no-new-privileges — blocks setuid/setgid escalation

If you cannot use containers, at minimum wrap the subprocess with systemd-run --scope --user --pty --pipe --wait --collect --service-type=exec -p MemoryMax=256M -p CPUQuota=50% -p TasksMax=50. The container approach is cleaner and more portable.

Step 4: configure the executor with a command allowlist

AutoGen’s LocalCommandLineCodeExecutor accepts a allowed_commands parameter. Restrict it to the bare minimum: python3 and perhaps pip if agents must install packages (prefer pre-baked images instead).

from autogen_ext.code_executors.local_command_line import LocalCommandLineCodeExecutor

executor = LocalCommandLineCodeExecutor(
    work_dir="/sandbox",
    allowed_commands=["python3"],
    timeout=30,
)

The allowed_commands list is matched against the first token of the command the agent tries to run. An agent sending python3 -c "import os; os.system('rm -rf /')" still passes the allowlist because the first token is python3. You need deeper inspection — see Step 5.

Also set a short timeout. Thirty seconds is generous for most agent tasks; drop to 10–15 seconds for interactive workloads.

Step 5: add a pre-execution validator

The allowlist is necessary but insufficient. Inject a validator that parses the agent’s proposed code before it reaches the executor. Reject anything that imports dangerous modules, uses subprocess, os.system, eval, exec, or accesses the network.

import ast
from typing import Set

DANGEROUS_IMPORTS: Set[str] = {
    "subprocess", "os", "sys", "shutil", "pathlib",
    "socket", "urllib", "requests", "http", "ftplib",
    "paramiko", "fabric", "invoke", "pexpect",
    "importlib", "pkgutil", "runpy", "zipimport",
    "ctypes", "cffi", "multiprocessing", "threading",
    "asyncio", "signal", "resource", "pwd", "grp",
    "crypt", "hashlib", "secrets", "hmac",
}

DANGEROUS_CALLS: Set[str] = {
    "eval", "exec", "compile", "execfile",
    "system", "popen", "spawn", "fork",
    "open", "read", "write", "unlink", "remove",
    "rmdir", "mkdir", "chdir", "chmod", "chown",
    "load", "loads", "dump", "dumps",
}

class CodeValidator(ast.NodeVisitor):
    def __init__(self) -> None:
        self.violations: list[str] = []

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            if alias.name.split(".")[0] in DANGEROUS_IMPORTS:
                self.violations.append(f"import {alias.name}")
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        if node.module and node.module.split(".")[0] in DANGEROUS_IMPORTS:
            self.violations.append(f"from {node.module} import ...")
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call) -> None:
        if isinstance(node.func, ast.Name) and node.func.id in DANGEROUS_CALLS:
            self.violations.append(f"call {node.func.id}()")
        elif isinstance(node.func, ast.Attribute) and node.func.attr in DANGEROUS_CALLS:
            self.violations.append(f"call .{node.func.attr}()")
        self.generic_visit(node)

def validate_code(code: str) -> list[str]:
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return [f"syntax error: {e}"]
    validator = CodeValidator()
    validator.visit(tree)
    return validator.violations

Wire it into your agent loop:

async def safe_execute(code: str) -> tuple[bool, str]:
    violations = validate_code(code)
    if violations:
        return False, f"rejected: {', '.join(violations)}"
    result = await executor.execute_code(code)
    return result.is_success, result.output or result.error

This catches static threats. It will not catch obfuscated payloads (getattr(__import__('os'), 'system')('id')) or runtime-generated code. Treat it as a filter, not a guarantee.

Step 6: enforce resource limits inside the sandbox

Container limits (Step 3) are your hard ceiling. Add soft limits inside the executor process so failures surface as structured errors rather than OOM kills.

import resource

def set_limits() -> None:
    # CPU time (seconds)
    resource.setrlimit(resource.RLIMIT_CPU, (10, 15))
    # Virtual memory (bytes) — 200 MiB
    resource.setrlimit(resource.RLIMIT_AS, (200 * 1024 * 1024, 256 * 1024 * 1024))
    # File size (bytes) — 10 MiB
    resource.setrlimit(resource.RLIMIT_FSIZE, (10 * 1024 * 1024, 10 * 1024 * 1024))
    # Open files
    resource.setrlimit(resource.RLIMIT_NOFILE, (32, 32))
    # Processes
    resource.setrlimit(resource.RLIMIT_NPROC, (10, 10))

# Call early in the executor entrypoint
set_limits()

Place this in a wrapper script that the container runs instead of the raw module:

# /usr/local/bin/autogen-executor-wrapper
import resource
import sys
from autogen_ext.code_executors.local_command_line import main

def set_limits() -> None:
    resource.setrlimit(resource.RLIMIT_CPU, (10, 15))
    resource.setrlimit(resource.RLIMIT_AS, (200 * 1024 * 1024, 256 * 1024 * 1024))
    resource.setrlimit(resource.RLIMIT_FSIZE, (10 * 1024 * 1024, 10 * 1024 * 1024))
    resource.setrlimit(resource.RLIMIT_NOFILE, (32, 32))
    resource.setrlimit(resource.RLIMIT_NPROC, (10, 10))

if __name__ == "__main__":
    set_limits()
    sys.exit(main())

Update the Dockerfile ENTRYPOINT to ["/usr/local/bin/autogen-executor-wrapper"] and copy the wrapper in.

Step 7: log every execution with structured metadata

You need an audit trail: what code ran, who requested it, how long it took, exit code, stdout/stderr length, and whether the validator flagged it. Emit JSON lines to stdout or a log shipper.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class ExecutionRecord:
    request_id: str
    timestamp: str
    agent_id: str
    code_hash: str
    code_length: int
    validator_violations: list[str]
    allowed: bool
    exit_code: Optional[int]
    duration_ms: int
    stdout_bytes: int
    stderr_bytes: int
    error: Optional[str]

async def execute_with_audit(
    code: str,
    agent_id: str,
    logger
) -> tuple[bool, str]:
    request_id = str(uuid.uuid4())
    timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    code_hash = hashlib.sha256(code.encode()).hexdigest()[:16]
    violations = validate_code(code)
    allowed = len(violations) == 0

    start = time.perf_counter()
    exit_code = None
    stdout_bytes = 0
    stderr_bytes = 0
    error = None
    output = ""

    if allowed:
        try:
            result = await executor.execute_code(code)
            exit_code = 0 if result.is_success else 1
            output = result.output or result.error or ""
            stdout_bytes = len((result.output or "").encode())
            stderr_bytes = len((result.error or "").encode())
            if not result.is_success:
                error = result.error
        except Exception as e:
            exit_code = -1
            error = str(e)
    else:
        output = f"rejected: {', '.join(violations)}"
        error = output

    duration_ms = int((time.perf_counter() - start) * 1000)

    record = ExecutionRecord(
        request_id=request_id,
        timestamp=timestamp,
        agent_id=agent_id,
        code_hash=code_hash,
        code_length=len(code),
        validator_violations=violations,
        allowed=allowed,
        exit_code=exit_code,
        duration_ms=duration_ms,
        stdout_bytes=stdout_bytes,
        stderr_bytes=stderr_bytes,
        error=error,
    )
    logger.info(json.dumps(asdict(record)))

    return allowed and exit_code == 0, output

Ship these logs to your observability stack (Datadog, Splunk, Loki, Elastic). Alert on:

• Rejected executions (validator violations)
• Non-zero exit codes
• Duration > 5 seconds
• Output size > 100 KB
• Same code_hash repeating (possible loop)

Step 8: test the hardening end to end

Write a test suite that attempts escapes and verifies they fail. Run it in CI on every change.

import pytest
from your_module import execute_with_audit, validate_code

class TestExecutorHardening:
    @pytest.mark.asyncio
    async def test_rejects_subprocess_import(self):
        code = "import subprocess; subprocess.run(['id'])"
        violations = validate_code(code)
        assert any("subprocess" in v for v in violations)

    @pytest.mark.asyncio
    async def test_rejects_os_system(self):
        code = "import os; os.system('id')"
        violations = validate_code(code)
        assert any("os" in v for v in violations)

    @pytest.mark.asyncio
    async def test_rejects_eval(self):
        code = "eval('1+1')"
        violations = validate_code(code)
        assert any("eval" in v for v in violations)

    @pytest.mark.asyncio
    async def test_rejects_file_write(self):
        code = "open('/etc/passwd', 'w').write('hacked')"
        violations = validate_code(code)
        assert any("open" in v for v in violations)

    @pytest.mark.asyncio
    async def test_allows_safe_math(self):
        code = "result = 2 ** 10; print(result)"
        violations = validate_code(code)
        assert violations == []
        success, output = await execute_with_audit(code, "test-agent", logger)
        assert success
        assert "1024" in output

    @pytest.mark.asyncio
    async def test_timeout_enforced(self):
        code = "import time; time.sleep(60)"
        success, output = await execute_with_audit(code, "test-agent", logger)
        assert not success
        assert "timeout" in output.lower() or "killed" in output.lower()

    @pytest.mark.asyncio
    async def test_memory_limit_enforced(self):
        code = "x = 'a' * (300 * 1024 * 1024)"  # 300 MB
        success, output = await execute_with_audit(code, "test-agent", logger)
        assert not success

Run the containerized executor against these tests:

docker run --rm \
  -v $(pwd)/tests:/tests:ro \
  autogen-executor:latest \
  pytest /tests/test_hardening.py -v

All tests must pass. If any escape succeeds, treat it as a critical vulnerability.

Step 9: rotate the sandbox and executor regularly

Even with hardening, assume compromise. Rotate the container daily (or per session for high-risk workloads). Use a sidecar or orchestrator to:

1. Stop the running executor container
2. Remove its tmpfs sandbox (automatic on stop)
3. Pull the latest executor image (rebuild on base image updates)
4. Start a fresh container
5. Update service discovery / load balancer

Kubernetes example with a CronJob and Deployment rollout:

# executor-rotation-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: autogen-executor-rotate
spec:
  schedule: "0 3 * * *"  # daily at 03:00 UTC
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: executor-rotator
          containers:
          - name: rotator
            image: bitnami/kubectl:latest
            command:
            - /bin/sh
            - -c
            - |
              kubectl rollout restart deployment/autogen-executor
              kubectl rollout status deployment/autogen-executor --timeout=5m
          restartPolicy: OnFailure

The executor-rotator SA needs patch/update on the deployment. This ensures zero-downtime rotation with fresh tmpfs each cycle.

Step 10: verify success in production

After deploying, confirm the hardening works under real traffic:

1. Check logs — Query your log store for validator_violations arrays with entries. You should see rejected attempts within hours.
2. Metric baselines — Establish p99 latency, error rate, and rejection rate. Alert on deviations > 2x.
3. Penetration test — Run a scoped internal test: give a red-team engineer agent access and measure time-to-escape. Document findings and close gaps.
4. Chaos test — Kill the executor container mid-request. Verify the agent receives a clean error and the sandbox is clean on restart.
5. Dependency scan — Run trivy or grype on the executor image weekly. Pin base image digests in the Dockerfile.

FROM python:3.11-slim@sha256:abc123...  # pinned digest

Closing notes

The LocalCommandLineCodeExecutor is a loaded gun. The steps above — dedicated user, container isolation, allowlist, static validator, resource limits, structured audit logs, automated tests, and daily rotation — form a defensible baseline. None of them alone is sufficient. Layer them.

If your workload allows, prefer the DockerCommandLineCodeExecutor or KubernetesCommandLineCodeExecutor (community-maintained) which move execution entirely out of the agent process. They are harder to operate but easier to secure.

For teams running inference gateways that route agent traffic across providers, the same principles apply: isolate execution, log everything, rotate aggressively. The gateway can enforce per-tenant executor pools with distinct resource quotas and audit sinks.

Ship the hardening first. Add agent capabilities second.