feat(code_executors): add opt-in sandbox flag to LocalCommandLineCodeExecutor (#7462)

Adds a ``sandbox: Optional[bool]`` parameter (plus ``sandbox_memory_bytes``) to
LocalCommandLineCodeExecutor and its Config, addressing #7462. Three modes:

- ``None`` (default): emits DeprecationWarning + logger.warning that a future
  release will make the parameter required. Execution unchanged -- fully
  backward compatible. Replaces the previous UserWarning.
- ``False``: explicit acknowledgement of unsandboxed execution; no warning.
- ``True``: best-effort in-process hardening on POSIX via ``preexec_fn``
  applying ``RLIMIT_CPU`` (timeout + 5s) and ``RLIMIT_AS`` (default 512 MiB),
  plus credential env-var scrub (``*_API_KEY``, ``*_TOKEN``, ``*_SECRET``,
  ``AWS_*``, ``OPENAI_*``, ``ANTHROPIC_*``, ``GITHUB_TOKEN``, etc.) applied
  to both code-execution and pip-install subprocesses. On Windows, resource
  limits are unavailable and a warning is logged; env scrub still applies.

Docstring is explicit that this is NOT a substitute for
DockerCommandLineCodeExecutor. This PR is draft pending maintainer direction
on Windows strategy before further investment.

Supersedes #7467 with broader scope.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: xr843

python/packages/autogen-ext/src/autogen_ext/code_executors/local/__init__.py

@@ -4,14 +4,15 @@
 import asyncio
 import logging
 import os
+import re
 import sys
 import tempfile
 import warnings
 from hashlib import sha256
 from pathlib import Path
 from string import Template
 from types import SimpleNamespace
-from typing import Any, Callable, ClassVar, List, Optional, Sequence, Union
+from typing import Any, Callable, ClassVar, Dict, List, Optional, Sequence, Tuple, Union
 
 from autogen_core import CancellationToken, Component
 from autogen_core.code_executor import CodeBlock, CodeExecutor, FunctionWithRequirements, FunctionWithRequirementsStr
@@ -32,6 +33,71 @@
 
 A = ParamSpec("A")
 
+logger = logging.getLogger(__name__)
+
+# Default hard memory cap applied via RLIMIT_AS when sandbox=True (POSIX only).
+# 512 MiB is enough for most small Python scripts while still bounding runaway
+# LLM-generated code. Callers that need more can override via the
+# ``sandbox_memory_bytes`` constructor argument.
+_DEFAULT_SANDBOX_MEMORY_BYTES: int = 512 * 1024 * 1024
+
+# Env-var name patterns scrubbed when sandbox=True. Matched case-insensitively
+# against variable *names* (not values). Best-effort only: this is not a
+# substitute for a real sandbox, and secrets stored under non-matching names
+# will still leak to the subprocess.
+_CREDENTIAL_ENV_PATTERNS: Tuple[re.Pattern[str], ...] = (
+    re.compile(r".*_API_KEY$", re.IGNORECASE),
+    re.compile(r".*_TOKEN$", re.IGNORECASE),
+    re.compile(r".*_SECRET$", re.IGNORECASE),
+    re.compile(r".*PASSWORD.*", re.IGNORECASE),
+    re.compile(r"^AWS_.*", re.IGNORECASE),
+    re.compile(r"^AZURE_.*", re.IGNORECASE),
+    re.compile(r"^GCP_.*", re.IGNORECASE),
+    re.compile(r"^GOOGLE_.*", re.IGNORECASE),
+    re.compile(r"^OPENAI_.*", re.IGNORECASE),
+    re.compile(r"^ANTHROPIC_.*", re.IGNORECASE),
+    re.compile(r"^HF_.*", re.IGNORECASE),
+    re.compile(r"^HUGGINGFACE_.*", re.IGNORECASE),
+    re.compile(r"^GITHUB_TOKEN$", re.IGNORECASE),
+    re.compile(r"^GH_TOKEN$", re.IGNORECASE),
+    re.compile(r"^NPM_TOKEN$", re.IGNORECASE),
+    re.compile(r"^PYPI_.*", re.IGNORECASE),
+)
+
+
+def _scrub_credentials_from_env(env: Dict[str, str]) -> Dict[str, str]:
+    """Return a copy of ``env`` with credential-shaped variable names removed."""
+    scrubbed: Dict[str, str] = {}
+    for key, value in env.items():
+        if any(pattern.match(key) for pattern in _CREDENTIAL_ENV_PATTERNS):
+            continue
+        scrubbed[key] = value
+    return scrubbed
+
+
+def _make_sandbox_preexec_fn(cpu_seconds: int, memory_bytes: int) -> Optional[Callable[[], None]]:
+    """Build a ``preexec_fn`` that applies POSIX resource limits before ``exec``.
+
+    Returns ``None`` on platforms where the ``resource`` module is unavailable
+    (notably Windows). The returned closure is intended to be passed as
+    ``preexec_fn`` to :func:`asyncio.create_subprocess_exec`.
+    """
+    try:
+        import resource  # type: ignore[import-not-found]
+    except ImportError:
+        return None
+
+    def _apply_limits() -> None:  # pragma: no cover - runs in child process
+        resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))
+        try:
+            resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes))
+        except (ValueError, OSError):
+            # Some platforms (e.g. macOS) refuse RLIMIT_AS; continue so
+            # sandbox=True still provides env-scrub + CPU capping.
+            pass
+
+    return _apply_limits
+
 
 class LocalCommandLineCodeExecutorConfig(BaseModel):
     """Configuration for LocalCommandLineCodeExecutor"""
@@ -40,6 +106,8 @@ class LocalCommandLineCodeExecutorConfig(BaseModel):
     work_dir: Optional[str] = None
     functions_module: str = "functions"
     cleanup_temp_files: bool = True
+    sandbox: Optional[bool] = None
+    sandbox_memory_bytes: int = _DEFAULT_SANDBOX_MEMORY_BYTES
 
 
 class LocalCommandLineCodeExecutor(CodeExecutor, Component[LocalCommandLineCodeExecutorConfig]):
@@ -81,6 +149,25 @@ class LocalCommandLineCodeExecutor(CodeExecutor, Component[LocalCommandLineCodeE
         functions_module (str, optional): The name of the module that will be created to store the functions. Defaults to "functions".
         cleanup_temp_files (bool, optional): Whether to automatically clean up temporary files after execution. Defaults to True.
         virtual_env_context (Optional[SimpleNamespace], optional): The virtual environment context. Defaults to None.
+        sandbox (Optional[bool], optional): Opt-in best-effort hardening for LLM-generated code.
+            This is **NOT** a substitute for :class:`~autogen_ext.code_executors.docker.DockerCommandLineCodeExecutor`
+            and provides no strong isolation guarantees; it is only a defense-in-depth layer for
+            users who cannot run Docker.
+
+            * ``None`` (default): backward-compatible behavior. Emits a :class:`DeprecationWarning`
+              noting that a future release will make this parameter required, and executes code
+              without any added hardening.
+            * ``False``: explicit acknowledgement that no sandboxing should be applied. No
+              warning is emitted.
+            * ``True``: applies POSIX ``RLIMIT_CPU`` (= ``timeout + 5`` seconds) and
+              ``RLIMIT_AS`` (default 512 MiB, see ``sandbox_memory_bytes``) via ``preexec_fn``
+              on code-execution and pip-install subprocesses, and scrubs credential-shaped
+              environment variables (``*_API_KEY``, ``*_TOKEN``, ``*_SECRET``, ``AWS_*``,
+              ``OPENAI_*``, etc.) from the subprocess environment. On Windows, resource limits
+              are unavailable and a warning is logged; env scrub still applies.
+        sandbox_memory_bytes (int, optional): Address-space cap (``RLIMIT_AS``) applied when
+            ``sandbox=True`` on POSIX. Defaults to 512 MiB. Ignored when ``sandbox`` is ``None``
+            or ``False``, and on platforms without the ``resource`` module.
 
     .. note::
         Using the current directory (".") as working directory is deprecated. Using it will raise a deprecation warning.
@@ -158,15 +245,38 @@ def __init__(
         functions_module: str = "functions",
         cleanup_temp_files: bool = True,
         virtual_env_context: Optional[SimpleNamespace] = None,
+        sandbox: Optional[bool] = None,
+        sandbox_memory_bytes: int = _DEFAULT_SANDBOX_MEMORY_BYTES,
     ):
-        # Issue warning about using LocalCommandLineCodeExecutor
-        warnings.warn(
-            "Using LocalCommandLineCodeExecutor may execute code on the local machine which can be unsafe. "
-            "For security, it is recommended to use DockerCommandLineCodeExecutor instead. "
-            "To install Docker, visit: https://docs.docker.com/get-docker/",
-            UserWarning,
-            stacklevel=2,
-        )
+        # Warn based on the caller's choice of ``sandbox``. When ``sandbox`` is
+        # left at its default (``None``), we emit a ``DeprecationWarning`` so
+        # callers are nudged toward making an explicit decision before a future
+        # release makes the parameter required. An explicit ``False`` opts out
+        # of the warning (but preserves the insecure behavior), while ``True``
+        # enables best-effort hardening. See issue #7462.
+        if sandbox is None:
+            deprecation_msg = (
+                "LocalCommandLineCodeExecutor was constructed without an explicit "
+                "`sandbox` argument. A future release will require this parameter. "
+                "Pass `sandbox=True` for best-effort in-process hardening (POSIX "
+                "resource limits + credential env-var scrub), or `sandbox=False` to "
+                "explicitly acknowledge unsandboxed execution. For real isolation, "
+                "prefer DockerCommandLineCodeExecutor. See "
+                "https://github.com/microsoft/autogen/issues/7462."
+            )
+            warnings.warn(deprecation_msg, DeprecationWarning, stacklevel=2)
+            logger.warning(deprecation_msg)
+
+        self._sandbox: Optional[bool] = sandbox
+        self._sandbox_memory_bytes: int = sandbox_memory_bytes
+        if sandbox is True and sys.platform == "win32":
+            logger.warning(
+                "LocalCommandLineCodeExecutor(sandbox=True) on Windows: POSIX "
+                "resource limits (RLIMIT_CPU, RLIMIT_AS) are not available on "
+                "this platform. Credential env-var scrubbing still applies, but "
+                "no CPU/memory caps will be enforced. Use "
+                "DockerCommandLineCodeExecutor for real isolation."
+            )
 
         if timeout < 1:
             raise ValueError("Timeout must be greater than or equal to 1.")
@@ -270,6 +380,33 @@ def cleanup_temp_files(self) -> bool:
         """(Experimental) Whether to automatically clean up temporary files after execution."""
         return self._cleanup_temp_files
 
+    @property
+    def sandbox(self) -> Optional[bool]:
+        """(Experimental) Whether sandbox hardening is enabled. See ``__init__`` docs."""
+        return self._sandbox
+
+    def _prepare_sandbox_subprocess_kwargs(self, env: Dict[str, str]) -> Tuple[Dict[str, str], Dict[str, Any]]:
+        """Apply sandbox policy to a subprocess env + kwargs pair.
+
+        Returns ``(env, extra_kwargs)``. When ``sandbox`` is not ``True``, the
+        env is returned unchanged and extra_kwargs is empty. When ``sandbox`` is
+        ``True``, credential-shaped env vars are stripped and (on POSIX) a
+        ``preexec_fn`` applying RLIMIT_CPU and RLIMIT_AS is attached.
+        """
+        if self._sandbox is not True:
+            return env, {}
+
+        scrubbed = _scrub_credentials_from_env(env)
+        extra_kwargs: Dict[str, Any] = {}
+        if sys.platform != "win32":
+            preexec = _make_sandbox_preexec_fn(
+                cpu_seconds=self._timeout + 5,
+                memory_bytes=self._sandbox_memory_bytes,
+            )
+            if preexec is not None:
+                extra_kwargs["preexec_fn"] = preexec
+        return scrubbed, extra_kwargs
+
     async def _setup_functions(self, cancellation_token: CancellationToken) -> None:
         func_file_content = build_python_functions_file(self._functions)
         func_file = self.work_dir / f"{self._functions_module}.py"
@@ -290,13 +427,17 @@ async def _setup_functions(self, cancellation_token: CancellationToken) -> None:
             else:
                 py_executable = sys.executable
 
+            pip_env, pip_extra_kwargs = self._prepare_sandbox_subprocess_kwargs(os.environ.copy())
+
             task = asyncio.create_task(
                 asyncio.create_subprocess_exec(
                     py_executable,
                     *cmd_args,
                     cwd=self.work_dir,
                     stdout=asyncio.subprocess.PIPE,
                     stderr=asyncio.subprocess.PIPE,
+                    env=pip_env,
+                    **pip_extra_kwargs,
                 )
             )
             cancellation_token.link_future(task)
@@ -422,6 +563,9 @@ async def _execute_code_dont_check_setup(
                     # Shell commands (bash, sh, etc.)
                     extra_args = [str(written_file.absolute())]
 
+            # Apply sandbox policy (env scrub + POSIX rlimits) if enabled.
+            env, sandbox_kwargs = self._prepare_sandbox_subprocess_kwargs(env)
+
             # Create a subprocess and run
             task = asyncio.create_task(
                 asyncio.create_subprocess_exec(
@@ -431,6 +575,7 @@ async def _execute_code_dont_check_setup(
                     stdout=asyncio.subprocess.PIPE,
                     stderr=asyncio.subprocess.PIPE,
                     env=env,
+                    **sandbox_kwargs,
                 )
             )
             cancellation_token.link_future(task)
@@ -514,6 +659,8 @@ def _to_config(self) -> LocalCommandLineCodeExecutorConfig:
             work_dir=str(self.work_dir),
             functions_module=self._functions_module,
             cleanup_temp_files=self._cleanup_temp_files,
+            sandbox=self._sandbox,
+            sandbox_memory_bytes=self._sandbox_memory_bytes,
         )
 
     @classmethod
@@ -523,4 +670,6 @@ def _from_config(cls, config: LocalCommandLineCodeExecutorConfig) -> Self:
             work_dir=Path(config.work_dir) if config.work_dir is not None else None,
             functions_module=config.functions_module,
             cleanup_temp_files=config.cleanup_temp_files,
+            sandbox=config.sandbox,
+            sandbox_memory_bytes=config.sandbox_memory_bytes,
         )

python/packages/autogen-ext/tests/code_executors/test_local_sandbox.py

@@ -0,0 +1,123 @@
+# Tests for the opt-in ``sandbox`` parameter on LocalCommandLineCodeExecutor.
+# See https://github.com/microsoft/autogen/issues/7462 for context.
+
+import os
+import sys
+import warnings
+
+import pytest
+from autogen_core import CancellationToken
+from autogen_core.code_executor import CodeBlock
+from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
+from autogen_ext.code_executors.local import (
+    LocalCommandLineCodeExecutorConfig,
+)
+
+
+def test_sandbox_none_emits_deprecation_warning() -> None:
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        LocalCommandLineCodeExecutor()
+
+    deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)]
+    user_warnings = [
+        w
+        for w in caught
+        if issubclass(w.category, UserWarning) and "DockerCommandLineCodeExecutor" in str(w.message)
+    ]
+    assert deprecations, "sandbox=None must emit a DeprecationWarning"
+    assert not user_warnings, "legacy UserWarning should be replaced by DeprecationWarning when sandbox=None"
+
+
+def test_sandbox_false_emits_no_warning() -> None:
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        LocalCommandLineCodeExecutor(sandbox=False)
+
+    sandbox_related = [
+        w
+        for w in caught
+        if "sandbox" in str(w.message).lower() or issubclass(w.category, DeprecationWarning)
+    ]
+    assert not sandbox_related, f"sandbox=False should not emit warnings, got: {[str(w.message) for w in caught]}"
+
+
+def test_sandbox_true_no_deprecation_warning() -> None:
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        LocalCommandLineCodeExecutor(sandbox=True)
+
+    deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)]
+    assert not deprecations, "sandbox=True should not emit DeprecationWarning"
+
+
+def test_config_round_trip_preserves_sandbox() -> None:
+    for sandbox_value in (None, True, False):
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            executor = LocalCommandLineCodeExecutor(sandbox=sandbox_value, sandbox_memory_bytes=123456789)
+        config = executor._to_config()
+        assert config.sandbox == sandbox_value
+        assert config.sandbox_memory_bytes == 123456789
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            rebuilt = LocalCommandLineCodeExecutor._from_config(config)
+        assert rebuilt.sandbox == sandbox_value
+        assert rebuilt._sandbox_memory_bytes == 123456789
+
+
+def test_config_model_defaults() -> None:
+    cfg = LocalCommandLineCodeExecutorConfig()
+    assert cfg.sandbox is None
+    assert cfg.sandbox_memory_bytes == 512 * 1024 * 1024
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only: env scrub + rlimits")
+@pytest.mark.asyncio
+async def test_sandbox_true_scrubs_credential_env_vars(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    monkeypatch.setenv("FAKE_API_KEY", "xyz-should-not-leak")
+    monkeypatch.setenv("HARMLESS_VAR", "keep-me")
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        executor = LocalCommandLineCodeExecutor(work_dir=tmp_path, sandbox=True)
+
+    result = await executor.execute_code_blocks(
+        code_blocks=[
+            CodeBlock(
+                language="bash",
+                code='echo "API=${FAKE_API_KEY:-MISSING}"; echo "OK=${HARMLESS_VAR:-MISSING}"',
+            )
+        ],
+        cancellation_token=CancellationToken(),
+    )
+    assert result.exit_code == 0, result.output
+    assert "API=MISSING" in result.output, f"FAKE_API_KEY was not scrubbed. Output: {result.output!r}"
+    assert "OK=keep-me" in result.output, f"HARMLESS_VAR should have been preserved. Output: {result.output!r}"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only")
+@pytest.mark.asyncio
+async def test_sandbox_false_preserves_env_vars(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    monkeypatch.setenv("FAKE_API_KEY", "xyz-should-be-visible")
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        executor = LocalCommandLineCodeExecutor(work_dir=tmp_path, sandbox=False)
+
+    result = await executor.execute_code_blocks(
+        code_blocks=[CodeBlock(language="bash", code='echo "API=${FAKE_API_KEY:-MISSING}"')],
+        cancellation_token=CancellationToken(),
+    )
+    assert result.exit_code == 0, result.output
+    assert "API=xyz-should-be-visible" in result.output
+
+
+def test_sandbox_true_imports_cleanly_on_windows() -> None:
+    # Smoke test: on any platform, constructing with sandbox=True must not raise.
+    # On Windows the Windows degrade path only logs a warning.
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        executor = LocalCommandLineCodeExecutor(sandbox=True)
+    assert executor.sandbox is True