fix(sandbox): keep document and seed values out of host shell code

src/agent_env/a2a_agent/a2a_agent.py

@@ -305,7 +305,7 @@ async def install_cli(
 
         if isinstance(sandbox, VmSandbox):
             await sandbox.exec_script(
-                f"docker exec -u 0 {shlex.quote(container or sandbox.container_name)} chmod +x {entrypoint_path}")
+                f"docker exec -u 0 {shlex.quote(container or sandbox.container_name)} chmod +x {shlex.quote(entrypoint_path)}")
         else:
             await sandbox.exec("chmod", "+x", entrypoint_path)
 
@@ -661,7 +661,7 @@ async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[
         env_flags = []
         for key, value in agent_env.items():
             escaped_value = value.replace("'", "'\\''")
-            env_flags.append(f"-e {key}='{escaped_value}'")
+            env_flags.append(f"-e {shlex.quote(key)}='{escaped_value}'")
         env_str = " \\\n    ".join(env_flags)
         publish = " ".join(
             f"-p {spec}" for spec in port_bindings(self._sandbox.host_ips, self._sandbox.host_port(a2a_port), a2a_port)
@@ -670,11 +670,11 @@ async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[
         run_script = f"""#!/bin/bash
 set -e
 {setup_script}docker {"run -d" if trust_dir is None else "create"} \\
-    --name {self._sandbox.container_name} \\
+    --name {shlex.quote(self._sandbox.container_name)} \\
     --label {shlex.quote(f"{SANDBOX_LABEL}={self._sandbox.sandbox_id}")} \\
     {publish} \\
     {network_flag}{env_str} \\
-    {image_name} > /dev/null
+    {shlex.quote(image_name)} > /dev/null
 {"sleep 2" if trust_dir is None else ""}
 """
         await self._sandbox.exec_script(run_script)

src/agent_env/providers/sandbox_providers/sandbox.py

@@ -476,7 +476,7 @@ async def _load_docker_images(self, artifacts: list, signed_urls: list[str | Non
                 # Download to a file first (retry-safe with -o); a `curl | ... docker load`
                 # pipe can't be retried without corrupting the stream (curl won't rewind).
                 load_commands.append(
-                    f'(curl -fsSL {CURL_RETRY_FLAGS} "{signed}" -o {shlex.quote(tmp_tar)} '
+                    f"(curl -fsSL {CURL_RETRY_FLAGS} {shlex.quote(signed)} -o {shlex.quote(tmp_tar)} "
                     f"&& gunzip -c {shlex.quote(tmp_tar)} | docker load && rm -f {shlex.quote(tmp_tar)})"
                 )
             else:
@@ -757,7 +757,7 @@ async def upload_vm_file(sandbox: VmSandbox, vm_path: str, store: ObjectStore, o
     machine."""
     put_url = await asyncio.to_thread(store.signed_put_url, object_url)
     if put_url is not None:
-        await sandbox.exec_script(f'curl -fsSL -X PUT --upload-file {shlex.quote(vm_path)} "{put_url}"')
+        await sandbox.exec_script(f"curl -fsSL -X PUT --upload-file {shlex.quote(vm_path)} {shlex.quote(put_url)}")
         return
     with tempfile.TemporaryDirectory(prefix="agentenv-vm-file-") as directory:
         local_path = os.path.join(directory, posixpath.basename(vm_path))

src/agent_env/task/task.py

@@ -9,6 +9,7 @@
 import itertools
 import json
 import logging
+import re
 import time
 import uuid
 from dataclasses import dataclass
@@ -94,18 +95,17 @@ def _overlapping_survivors(
 
 def substitute_seed_in_steps(steps: list["TaskStep"], seed: dict) -> list["TaskStep"]:
     """Round-trip each step through its own to_dict/from_dict with ``<key>``
-    seed placeholders replaced in every string leaf. Returns a NEW list, never
-    mutating the input; idempotent (an already-substituted string has no tokens
-    left). Used by Task.run(); a retry re-dispatches the same substituted step
-    objects, so it executes exactly what the original run executed."""
+    seed placeholders replaced in every string leaf, in one left-to-right pass
+    so substituted text is never re-scanned. Returns a NEW list, never mutating
+    the input. Used by Task.run(); a retry re-dispatches the same substituted
+    step objects, so it executes exactly what the original run executed."""
     if not seed:
         return list(steps)
+    placeholders = re.compile("|".join(re.escape(f"<{key}>") for key in seed))
 
     def _sub(value):
         if isinstance(value, str):
-            for k, v in seed.items():
-                value = value.replace(f"<{k}>", str(v))
-            return value
+            return placeholders.sub(lambda match: str(seed[match.group(0)[1:-1]]), value)
         if isinstance(value, list):
             return [_sub(x) for x in value]
         if isinstance(value, dict):
@@ -488,8 +488,7 @@ async def _run(
         # and concurrent runs must not race on a shared rewrite. And applied on
         # EVERY start_step, not only 0: a resume reloads the task with its
         # placeholders intact, so skipping substitution would hand later steps
-        # literal `<multienv_id>` strings. Substitution is idempotent — an
-        # already-replaced string has no tokens left.
+        # literal `<multienv_id>` strings.
         seed = context.metadata.get("seed") or {}
         steps = substitute_seed_in_steps(self.steps, seed) if seed else self.steps
         if start_step < 0 or start_step >= len(steps):

src/agent_env/task_step/task_steps/run_docker_container.py

@@ -196,7 +196,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         dockerfile_abs = posixpath.join(work_dir, self.dockerfile_path)
         await sandbox.exec_script(
             f"test -f {shlex.quote(dockerfile_abs)} "
-            f"|| (echo 'Dockerfile not found at {self.dockerfile_path} inside the build context' >&2; exit 1)"
+            f"|| (echo {shlex.quote(f'Dockerfile not found at {self.dockerfile_path} inside the build context')} >&2; "
+            f"exit 1)"
         )
 
         build_arg_flags = ""
@@ -286,7 +287,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 f"for i in $(seq 1 60); do "
                 f"if docker exec {shlex.quote(container)} sh -c {ready} >/dev/null 2>&1; then exit 0; fi; "
                 f"sleep 2; done; "
-                f"echo 'container {self.container_name} failed readiness check within 120s' >&2; exit 1"
+                f"echo {shlex.quote(f'container {self.container_name} failed readiness check within 120s')} >&2; exit 1"
             )
 
         context.metadata.setdefault("deployed_docker_containers", []).append({

src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py

@@ -180,14 +180,11 @@ def _resolve_command(self, context: TaskStepContext) -> tuple[str, dict[str, str
         new task version: ``step_params`` for a targeted or resumed run, ``seed`` for
         ``task run-batch --seeds`` fan-out. Both are additive to the stored ``env_vars``.
 
-        Seed values reach the container as env vars rather than by string substitution,
-        because ``-e K=<quoted>`` is a separate docker argument and bash expands ``"$K"``
-        without re-parsing metacharacters. ``<key>`` placeholders are still rendered for
-        rows that vary a flag rather than a value, in one left-to-right pass so
-        substituted text is never re-scanned: replacing key-by-key, a value of ``<b>``
-        quotes to ``'<b>'`` and a later key ``b`` rewrites the placeholder *inside* those
-        quotes, yielding ``''$(id)''`` — quotes that close each other and leave the
-        value bare for bash.
+        Seed values reach the container as env vars, ``-e K=<quoted>``, and bash expands
+        ``"$K"`` without re-parsing them, so a command that takes data reads it that way.
+        A ``<key>`` in the stored command was already filled with the value as written
+        when the run started (``substitute_seed_in_steps``); this pass shell-quotes the
+        placeholders of a ``step_params`` command only.
         """
         overrides = self.step_param_overrides(context)
         seed = context.metadata.get("seed") or {}
@@ -432,7 +429,7 @@ async def _run_in_container(self, sandbox, merged_env: dict[str, str],
         )
         wrapped = (
             f"docker exec -u {user} {env_flags} {container} "
-            f"timeout --kill-after=10 {self.timeout_sec} bash -c {shlex.quote(command)}"
+            f"timeout --kill-after=10 {shlex.quote(str(self.timeout_sec))} bash -c {shlex.quote(command)}"
         )
         exit_code, stdout, stderr = await sandbox.exec_with_output("sudo", "bash", "-c", wrapped)
         return exit_code, stdout, stderr, await self._extracted(lambda path: self._extract_file(sandbox, path))

tst/unit/providers/sandbox_providers/local_sandbox_test.py

@@ -24,6 +24,7 @@
 from agent_env.a2a_agent import a2a_agent as a2a_agent_module
 from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from tst.unit.store.fakes import FakeObjectStore
+from tst.util.exec_scripts import run_with_stubs
 
 
 def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
@@ -1292,6 +1293,20 @@ async def test_an_agent_placed_on_a_local_vm_sandbox_maps_the_host_alias(tmp_pat
     assert "--add-host host.docker.internal:host-gateway" in sandbox.scripts[0]
 
 
+@pytest.mark.asyncio
+async def test_an_agent_env_var_whose_name_holds_shell_syntax_reaches_docker_as_it_is(tmp_path, monkeypatch):
+    monkeypatch.setattr(a2a_agent_module, "local_grant_trust", lambda: None)
+    sandbox = _RecordingLocalSandbox(work_dir=tmp_path / "work")
+    agent = A2AAgent.__new__(A2AAgent)
+    agent._sandbox = sandbox
+
+    await agent._run_container("img:v1", 8000, {"$(touch pwned)": "x';touch pwned;'"})
+
+    docker, = run_with_stubs(sandbox.scripts[0], tmp_path, "docker", "sleep")["docker"]
+    assert "$(touch pwned)=x';touch pwned;'" in docker and docker[-1] == "img:v1"
+    assert not (tmp_path / "pwned").exists()
+
+
 async def _no_sleep(_seconds):
     return None
 

tst/unit/providers/sandbox_providers/vm_file_transfer_test.py

@@ -11,6 +11,7 @@
 from agent_env.providers.sandbox_providers import sandbox as sandbox_module
 from agent_env.providers.sandbox_providers.sandbox import read_vm_file, upload_vm_file
 from agent_env.store.object_store.local import LocalFilesystemObjectStore
+from tst.util.exec_scripts import run_with_stubs
 
 
 class _ShellVm:
@@ -104,7 +105,19 @@ async def test_a_store_that_signs_uploads_is_uploaded_to_by_the_vm(tmp_path):
 
     await upload_vm_file(vm, "/tmp/image.tar.gz", _SigningStore(), "s3://bucket/builds/image.tar.gz")
 
-    assert vm.scripts == ['curl -fsSL -X PUT --upload-file /tmp/image.tar.gz "https://put.example/image.tar.gz?sig=1"']
+    assert vm.scripts == ["curl -fsSL -X PUT --upload-file /tmp/image.tar.gz 'https://put.example/image.tar.gz?sig=1'"]
+
+
+@pytest.mark.asyncio
+async def test_a_presigned_url_with_shell_syntax_reaches_curl_as_it_is(tmp_path):
+    vm = _RecordingVm()
+    name = "a$(touch pwned)`touch pwned`;touch pwned"
+
+    await upload_vm_file(vm, "/tmp/image.tar.gz", _SigningStore(), f"s3://bucket/builds/{name}")
+
+    curl, = run_with_stubs(vm.scripts[0], tmp_path, "curl")["curl"]
+    assert curl[-1] == f"https://put.example/{name}?sig=1"
+    assert not (tmp_path / "pwned").exists()
 
 
 @pytest.mark.asyncio

tst/unit/providers/sandbox_providers/vm_sandbox_test.py

@@ -14,7 +14,7 @@
 from agent_env.store import ImageStore, RegistryAuth, set_object_store
 from agent_env.config import get_config, reset_config
 from tst.unit.store.fakes import ConfiguredObjectStore
-from tst.util.exec_scripts import script_run
+from tst.util.exec_scripts import run_with_stubs, script_run
 
 
 def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
@@ -131,13 +131,25 @@ async def test_a_load_of_tarballs_runs_the_script_it_always_has(signing_store):
 
     flags = "--retry 5 --retry-all-errors --retry-delay 1"
     assert sandbox.scripts == [
-        f'(curl -fsSL {flags} "https://signed/a.tar.gz" -o /tmp/_docker_image_vm-test_0.tar.gz '
+        f"(curl -fsSL {flags} https://signed/a.tar.gz -o /tmp/_docker_image_vm-test_0.tar.gz "
         "&& gunzip -c /tmp/_docker_image_vm-test_0.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_0.tar.gz) & "
-        f'(curl -fsSL {flags} "https://signed/b.tar.gz" -o /tmp/_docker_image_vm-test_1.tar.gz '
+        f"(curl -fsSL {flags} https://signed/b.tar.gz -o /tmp/_docker_image_vm-test_1.tar.gz "
         "&& gunzip -c /tmp/_docker_image_vm-test_1.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_1.tar.gz) & wait"
     ]
 
 
+@pytest.mark.asyncio
+async def test_a_signed_url_with_shell_syntax_reaches_curl_as_it_is(signing_store, tmp_path):
+    name = "a$(touch pwned)`touch pwned`;touch pwned;&.tar.gz"
+    sandbox = _RecordingVmSandbox(images_stdout="a\n")
+
+    await sandbox.load_docker_images([_image(f"s3://bucket/{name}", "a:1")])
+
+    curl, = run_with_stubs(sandbox.scripts[0], tmp_path, "curl", "gunzip", "docker")["curl"]
+    assert curl[-3:] == [f"https://signed/{name}", "-o", "/tmp/_docker_image_vm-test_0.tar.gz"]
+    assert not (tmp_path / "pwned").exists()
+
+
 class _Registry(ImageStore):
     """Holds credentials for registry.example only."""
 
@@ -177,7 +189,7 @@ async def test_tarballs_are_loaded_and_the_rest_pulled(signing_store, registry):
     await sandbox.load_docker_images([_image(None, PUBLIC), _image("s3://bucket/a.tar.gz", "a:1")])
 
     load, pull = sandbox.scripts
-    assert '"https://signed/a.tar.gz"' in load and PUBLIC not in load
+    assert " https://signed/a.tar.gz " in load and PUBLIC not in load
     assert pull == f"docker pull {PUBLIC}"
 
 
@@ -972,7 +984,7 @@ async def test_tarballs_are_loaded_registry_images_pulled_and_contexts_built(sig
     await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz", "a:1"), _image(None, PUBLIC),
                                       _context_image("c", "sha256:" + "3" * 64)])
 
-    assert '"https://signed/a.tar.gz"' in sandbox.scripts[0]
+    assert " https://signed/a.tar.gz " in sandbox.scripts[0]
     assert sandbox.scripts[1] == f"docker pull {PUBLIC}"
     assert not any("local/c-0123456789ab" in script for script in sandbox.scripts if script.startswith("docker pull"))
     assert sandbox.scripts[-2].endswith("-t local/c-0123456789ab:v1 .")

tst/unit/store/aws/conftest.py

@@ -6,10 +6,12 @@
 """
 
 import os
+import threading
 
 import pytest
 
 pytest.importorskip("boto3")
+botocore_client = pytest.importorskip("botocore.client")
 
 _PLACEHOLDER_AWS_ENV = {
     "AWS_ACCESS_KEY_ID": "testing",
@@ -25,3 +27,17 @@ def _placeholder_aws_credentials(monkeypatch):
         for name, value in _PLACEHOLDER_AWS_ENV.items():
             if name not in os.environ:
                 monkeypatch.setenv(name, value)
+
+
+@pytest.fixture
+def one_aws_call_at_a_time(monkeypatch):
+    """moto checks a write's condition and applies the write as two steps any thread can come
+    between, where DynamoDB applies each call atomically; one call at a time restores that for
+    tests that race writers."""
+    serial, make_api_call = threading.Lock(), botocore_client.BaseClient._make_api_call
+
+    def call(client, operation, params):
+        with serial:
+            return make_api_call(client, operation, params)
+
+    monkeypatch.setattr(botocore_client.BaseClient, "_make_api_call", call)

tst/unit/store/aws/dynamodb_document_store_test.py

@@ -1,8 +1,5 @@
 """DynamoDbDocumentStore runs the full DocumentStore conformance suite against moto."""
 
-import threading
-
-import botocore.client
 import pytest
 from moto import mock_aws
 
@@ -12,17 +9,7 @@
 
 
 @pytest.fixture
-def store_coll(monkeypatch):
-    """moto checks a write's condition and applies the write as two steps any thread can come
-    between, where DynamoDB applies each call atomically; one call at a time restores that for
-    the kit's racing cases."""
-    serial, make_api_call = threading.Lock(), botocore.client.BaseClient._make_api_call
-
-    def one_call_at_a_time(client, operation, params):
-        with serial:
-            return make_api_call(client, operation, params)
-
-    monkeypatch.setattr(botocore.client.BaseClient, "_make_api_call", one_call_at_a_time)
+def store_coll(one_aws_call_at_a_time):
     with mock_aws():
         yield DynamoDbDocumentStore(table_prefix="test_", region="us-west-2"), "coll"
 

tst/unit/store/aws/dynamodb_step_journal_test.py

@@ -14,7 +14,7 @@
 
 
 @pytest.fixture
-def store():
+def store(one_aws_call_at_a_time):
     with mock_aws():
         set_document_store(DynamoDbDocumentStore(table_prefix="test_", region="us-west-2"))
         s = TaskInstanceStore()

tst/unit/task/seed_substitution_test.py

@@ -0,0 +1,30 @@
+"""A run's seed fills each step's ``<key>`` placeholders in one left-to-right pass, so text a value brings in is
+never read again as another placeholder."""
+
+from __future__ import annotations
+
+from agent_env.task.task import substitute_seed_in_steps
+from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import (
+    RunContainerUnitTestsVerifierTaskStep,
+)
+
+
+def _verifier(**fields):
+    return RunContainerUnitTestsVerifierTaskStep(id="check", version=None, sandbox_name="h", container_name="c",
+                                                 **{"command": "true", **fields})
+
+
+def test_each_placeholder_takes_its_value():
+    step = _verifier(command='qc --task-id <task_id> --universe-id "<universe_id>" --turns <turns>')
+
+    seeded, = substitute_seed_in_steps([step], {"task_id": "task-12", "universe_id": "hg4/v3.1", "turns": 8})
+
+    assert seeded.command == 'qc --task-id task-12 --universe-id "hg4/v3.1" --turns 8'
+
+
+def test_substituted_text_is_never_rescanned():
+    step = _verifier(env_vars={"A": "<a>", "B": "<b>"})
+
+    seeded, = substitute_seed_in_steps([step], {"a": "<b>", "b": "$(id)"})
+
+    assert seeded.env_vars == {"A": "<b>", "B": "$(id)"}

tst/unit/task_step/local_step_containers_test.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import asyncio
+import subprocess
 import types
 from unittest.mock import AsyncMock
 
@@ -24,6 +25,7 @@
 from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import (
     RunContainerUnitTestsVerifierTaskStep,
 )
+from tst.util.exec_scripts import run_with_stubs
 
 SANDBOX_ID = "local-ab12cd34"
 HOST_PORT = 50123
@@ -115,6 +117,26 @@ async def test_a_tag_and_a_container_network_named_by_the_step_are_the_sandboxs_
     assert f"--network container:worker-{SANDBOX_ID} " in run
 
 
+@pytest.mark.asyncio
+async def test_a_dockerfile_path_and_container_name_holding_shell_syntax_run_nothing_when_reported(
+    tmp_path, monkeypatch,
+):
+    sandbox = _RecordingSandbox(tmp_path)
+    context = _on(monkeypatch, sandbox)
+    monkeypatch.setattr(RunDockerContainerTaskStep, "_stage_from_universe", AsyncMock())
+
+    await RunDockerContainerTaskStep(
+        id="box", version=None, sandbox_name="h", docker_context_artifact_id="u", docker_context_artifact_version=1,
+        dockerfile_path="missing'; touch pwned; #", container_name="c';touch pwned;'", ready_command="true",
+    ).execute(context)
+
+    reports = [s for s in sandbox.scripts if s.startswith("test -f ") or "failed readiness check" in s]
+    for script in reports:
+        with pytest.raises(subprocess.CalledProcessError):
+            run_with_stubs(script, tmp_path, "seq", "docker")
+    assert len(reports) == 2 and not (tmp_path / "pwned").exists()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("error, falls_back", [
     ("ERROR: failed to solve: no match for platform in manifest: not found", True),