Fail closed on unsafe code execution without explicit policy and pre-execution confirmation

lib/crewai/src/crewai/agent/core.py

@@ -186,6 +186,17 @@ class Agent(BaseAgent):
     allow_code_execution: bool | None = Field(
         default=False, description="Enable code execution for the agent."
     )
+    allow_unsafe_code_execution: bool = Field(
+        default=False,
+        description="Explicit policy opt-in required to allow unsafe code execution mode.",
+    )
+    unsafe_code_execution_confirmation: Any | None = Field(
+        default=None,
+        description=(
+            "Callable confirmation gate executed before unsafe code tools are enabled. "
+            "Must return True to allow execution."
+        ),
+    )
     respect_context_window: bool = Field(
         default=True,
         description="Keep messages under the context window size by summarizing content.",

lib/crewai/src/crewai/crew.py

@@ -1282,6 +1282,7 @@ def _prepare_tools(
         if hasattr(agent, "allow_code_execution") and getattr(
             agent, "allow_code_execution", False
         ):
+            self._validate_code_execution_safety_policy(agent, task)
             tools = self._add_code_execution_tools(agent, tools)
 
         if (
@@ -1410,9 +1411,40 @@ def _add_code_execution_tools(
             return self._merge_tools(tools, cast(list[BaseTool], code_tools))
         return tools
 
-    def _add_memory_tools(
-        self, tools: list[BaseTool], memory: Any
-    ) -> list[BaseTool]:
+    @staticmethod
+    def _validate_code_execution_safety_policy(agent: BaseAgent, task: Task) -> None:
+        code_execution_mode = getattr(agent, "code_execution_mode", "safe")
+        if code_execution_mode != "unsafe":
+            return
+
+        if not getattr(agent, "allow_unsafe_code_execution", False):
+            raise ValueError(
+                "Unsafe code execution is disabled by default. "
+                "Set allow_unsafe_code_execution=True to opt in."
+            )
+
+        confirmation_gate = getattr(agent, "unsafe_code_execution_confirmation", None)
+        if not callable(confirmation_gate):
+            raise ValueError(
+                "Unsafe code execution requires agent.unsafe_code_execution_confirmation "
+                "to be a callable that returns True before execution."
+            )
+
+        try:
+            confirmed = bool(confirmation_gate(task))
+        except TypeError:
+            confirmed = bool(confirmation_gate())
+        except Exception as exc:
+            raise ValueError(
+                "Unsafe code execution confirmation gate raised an exception."
+            ) from exc
+
+        if not confirmed:
+            raise ValueError(
+                "Unsafe code execution confirmation gate must return True."
+            )
+
+    def _add_memory_tools(self, tools: list[BaseTool], memory: Any) -> list[BaseTool]:
         """Add recall and remember tools when memory is available.
 
         Args:

lib/crewai/src/crewai/project/crew_base.py

@@ -75,6 +75,8 @@ class AgentConfig(TypedDict, total=False):
 
     # Code execution
     allow_code_execution: bool
+    allow_unsafe_code_execution: bool
+    unsafe_code_execution_confirmation: Any
     code_execution_mode: Literal["safe", "unsafe"]
 
     # Context and performance

lib/crewai/tests/test_crew.py

@@ -1655,6 +1655,113 @@ def test_code_execution_flag_adds_code_tool_upon_kickoff():
             )
 
 
+def test_unsafe_code_execution_requires_explicit_allow_policy():
+    with patch.object(Agent, "_validate_docker_installation"):
+        agent = Agent(
+            role="Programmer",
+            goal="Write code to solve problems.",
+            backstory="You're a programmer who loves to solve problems with code.",
+            allow_code_execution=True,
+            code_execution_mode="unsafe",
+        )
+
+    task = Task(
+        description="Write a script.",
+        expected_output="A working script.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task])
+
+    with pytest.raises(
+        ValueError, match="Set allow_unsafe_code_execution=True to opt in."
+    ):
+        crew.kickoff()
+
+
+def test_unsafe_code_execution_requires_confirmation_setting():
+    with patch.object(Agent, "_validate_docker_installation"):
+        agent = Agent(
+            role="Programmer",
+            goal="Write code to solve problems.",
+            backstory="You're a programmer who loves to solve problems with code.",
+            allow_code_execution=True,
+            allow_unsafe_code_execution=True,
+            code_execution_mode="unsafe",
+        )
+
+    task = Task(
+        description="Write a script.",
+        expected_output="A working script.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task])
+
+    with pytest.raises(
+        ValueError,
+        match="unsafe_code_execution_confirmation",
+    ):
+        crew.kickoff()
+
+
+def test_unsafe_code_execution_requires_positive_confirmation():
+    with patch.object(Agent, "_validate_docker_installation"):
+        agent = Agent(
+            role="Programmer",
+            goal="Write code to solve problems.",
+            backstory="You're a programmer who loves to solve problems with code.",
+            allow_code_execution=True,
+            allow_unsafe_code_execution=True,
+            unsafe_code_execution_confirmation=lambda *_args: False,
+            code_execution_mode="unsafe",
+        )
+
+    task = Task(
+        description="Write a script.",
+        expected_output="A working script.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task])
+
+    with pytest.raises(
+        ValueError,
+        match="confirmation gate must return True",
+    ):
+        crew.kickoff()
+
+
+def test_unsafe_code_execution_allowed_with_policy_and_confirmation_gate():
+    with patch.object(Agent, "_validate_docker_installation"):
+        agent = Agent(
+            role="Programmer",
+            goal="Write code to solve problems.",
+            backstory="You're a programmer who loves to solve problems with code.",
+            allow_code_execution=True,
+            allow_unsafe_code_execution=True,
+            unsafe_code_execution_confirmation=lambda *_args: True,
+            code_execution_mode="unsafe",
+        )
+
+    task = Task(
+        description="Write a script.",
+        expected_output="A working script.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task])
+    mock_task_output = TaskOutput(
+        description="Mock description", raw="mocked output", agent="mocked agent"
+    )
+
+    with patch.object(
+        Task, "execute_sync", return_value=mock_task_output
+    ) as mock_execute_sync:
+        crew.kickoff()
+        _, kwargs = mock_execute_sync.call_args
+        used_tools = kwargs["tools"]
+        assert any(isinstance(tool, CodeInterpreterTool) for tool in used_tools), (
+            "CodeInterpreterTool should be present"
+        )
+
+
 @pytest.mark.vcr()
 def test_delegation_is_not_enabled_if_there_are_only_one_agent():
     researcher = Agent(