diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 22b5a24ca1..e80faa8f1c 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/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.", diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 980830af52..40a583df1b 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/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: diff --git a/lib/crewai/src/crewai/project/crew_base.py b/lib/crewai/src/crewai/project/crew_base.py index 323450b139..1c76fcf176 100644 --- a/lib/crewai/src/crewai/project/crew_base.py +++ b/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 diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index 64d122a7ce..c53c7b0291 100644 --- a/lib/crewai/tests/test_crew.py +++ b/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(