fix: ignore expected security warnings in subprocess_utils.py

- Add per-file-ignores for S602/S603 in subprocess_utils.py
- These warnings are expected for Windows shell=True compatibility fix
- Resolves remaining CI lint failures for Windows subprocess fix

Co-Authored-By: João <joao@crewai.com>

pyproject.toml

@@ -135,6 +135,7 @@ ignore = ["E501"]  # ignore line too long
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = ["S101"]  # Allow assert statements in tests
+"src/crewai/cli/subprocess_utils.py" = ["S602", "S603"]  # Allow shell=True for Windows compatibility
 
 [tool.mypy]
 exclude = ["src/crewai/cli/templates", "tests"]

src/crewai/cli/evaluate_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def evaluate_crew(n_iterations: int, model: str) -> None:
     """
@@ -17,7 +19,7 @@ def evaluate_crew(n_iterations: int, model: str) -> None:
         if n_iterations <= 0:
             raise ValueError("The number of iterations must be a positive integer.")
 
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/git.py

@@ -1,6 +1,8 @@
 import subprocess
 from functools import lru_cache
 
+from crewai.cli.subprocess_utils import run_command
+
 
 class Repository:
     def __init__(self, path="."):
@@ -17,7 +19,7 @@ class Repository:
     def is_git_installed(self) -> bool:
         """Check if Git is installed and available in the system."""
         try:
-            subprocess.run(
+            run_command(
                 ["git", "--version"], capture_output=True, check=True, text=True
             )
             return True
@@ -26,24 +28,29 @@ class Repository:
 
     def fetch(self) -> None:
         """Fetch latest updates from the remote."""
-        subprocess.run(["git", "fetch"], cwd=self.path, check=True)
+        run_command(["git", "fetch"], cwd=self.path, check=True)
 
     def status(self) -> str:
         """Get the git status in porcelain format."""
-        return subprocess.check_output(
+        result = run_command(
             ["git", "status", "--branch", "--porcelain"],
             cwd=self.path,
-            encoding="utf-8",
-        ).strip()
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        return result.stdout.strip()
 
     @lru_cache(maxsize=None)
     def is_git_repo(self) -> bool:
         """Check if the current directory is a git repository."""
         try:
-            subprocess.check_output(
+            run_command(
                 ["git", "rev-parse", "--is-inside-work-tree"],
                 cwd=self.path,
-                encoding="utf-8",
+                capture_output=True,
+                text=True,
+                check=True,
             )
             return True
         except subprocess.CalledProcessError:
@@ -70,7 +77,7 @@ class Repository:
     def origin_url(self) -> str | None:
         """Get the Git repository's remote URL."""
         try:
-            result = subprocess.run(
+            result = run_command(
                 ["git", "remote", "get-url", "origin"],
                 cwd=self.path,
                 capture_output=True,

src/crewai/cli/install_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 # Be mindful about changing this.
 # on some environments we don't use this command but instead uv sync directly
@@ -12,8 +14,8 @@ def install_crew(proxy_options: list[str]) -> None:
     Install the crew by running the UV command to lock and install.
     """
     try:
-        command = ["uv", "sync"] + proxy_options
-        subprocess.run(command, check=True, capture_output=False, text=True)
+        command = ["uv", "sync", *proxy_options]
+        run_command(command, check=True, capture_output=False, text=True)
 
     except subprocess.CalledProcessError as e:
         click.echo(f"An error occurred while running the crew: {e}", err=True)

src/crewai/cli/kickoff_flow.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def kickoff_flow() -> None:
     """
@@ -10,7 +12,7 @@ def kickoff_flow() -> None:
     command = ["uv", "run", "kickoff"]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/plot_flow.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def plot_flow() -> None:
     """
@@ -10,7 +12,7 @@ def plot_flow() -> None:
     command = ["uv", "run", "plot"]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/replay_from_task.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def replay_task_command(task_id: str) -> None:
     """
@@ -13,7 +15,7 @@ def replay_task_command(task_id: str) -> None:
     command = ["uv", "run", "replay", task_id]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
         if result.stderr:
             click.echo(result.stderr, err=True)
 

src/crewai/cli/run_crew.py

@@ -1,12 +1,12 @@
 import subprocess
 from enum import Enum
-from typing import List, Optional
 
 import click
 from packaging import version
 
 from crewai.cli.utils import read_toml
 from crewai.cli.version import get_crewai_version
+from crewai.cli.subprocess_utils import run_command
 
 
 class CrewType(Enum):
@@ -57,7 +57,7 @@ def execute_command(crew_type: CrewType) -> None:
     command = ["uv", "run", "kickoff" if crew_type == CrewType.FLOW else "run_crew"]
 
     try:
-        subprocess.run(command, capture_output=False, text=True, check=True)
+        run_command(command, capture_output=False, text=True, check=True)
 
     except subprocess.CalledProcessError as e:
         handle_error(e, crew_type)

src/crewai/cli/subprocess_utils.py

@@ -0,0 +1,60 @@
+import platform
+import subprocess
+from typing import Any
+
+
+def run_command(
+    command: list[str],
+    capture_output: bool = False,
+    text: bool = True,
+    check: bool = True,
+    cwd: str | None = None,
+    env: dict[str, str] | None = None,
+    **kwargs: Any
+) -> subprocess.CompletedProcess:
+    """
+    Cross-platform subprocess execution with Windows compatibility.
+
+    On Windows, uses shell=True to avoid permission issues with restrictive
+    security policies. On other platforms, uses the standard approach.
+
+    Args:
+        command: List of command arguments
+        capture_output: Whether to capture stdout/stderr
+        text: Whether to use text mode
+        check: Whether to raise CalledProcessError on non-zero exit
+        cwd: Working directory
+        env: Environment variables
+        **kwargs: Additional subprocess.run arguments
+
+    Returns:
+        CompletedProcess instance
+
+    Raises:
+        subprocess.CalledProcessError: If check=True and command fails
+    """
+    if platform.system() == "Windows":
+        if isinstance(command, list):
+            command_str = subprocess.list2cmdline(command)
+        else:
+            command_str = command
+
+        return subprocess.run(
+            command_str,
+            shell=True,
+            capture_output=capture_output,
+            text=text,
+            check=check,
+            cwd=cwd,
+            env=env,
+            **kwargs
+        )
+    return subprocess.run(
+        command,
+        capture_output=capture_output,
+        text=text,
+        check=check,
+        cwd=cwd,
+        env=env,
+        **kwargs
+    )

src/crewai/cli/tools/main.py

@@ -1,6 +1,5 @@
 import base64
 import os
-import subprocess
 import tempfile
 from pathlib import Path
 from typing import Any
@@ -11,6 +10,7 @@ from rich.console import Console
 from crewai.cli import git
 from crewai.cli.command import BaseCommand, PlusAPIMixin
 from crewai.cli.config import Settings
+from crewai.cli.subprocess_utils import run_command
 from crewai.cli.utils import (
     extract_available_exports,
     get_project_description,
@@ -56,7 +56,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         os.chdir(project_root)
         try:
             self.login()
-            subprocess.run(["git", "init"], check=True)
+            run_command(["git", "init"], check=True)
             console.print(
                 f"[green]Created custom tool [bold]{folder_name}[/bold]. Run [bold]cd {project_root}[/bold] to start working.[/green]"
             )
@@ -94,7 +94,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         self._print_current_organization()
 
         with tempfile.TemporaryDirectory() as temp_build_dir:
-            subprocess.run(
+            run_command(
                 ["uv", "build", "--sdist", "--out-dir", temp_build_dir],
                 check=True,
                 capture_output=False,
@@ -196,7 +196,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         else:
             add_package_command.extend(["--index", index, tool_handle])
 
-        add_package_result = subprocess.run(
+        add_package_result = run_command(
             add_package_command,
             capture_output=False,
             env=self._build_env_with_credentials(repository_handle),

src/crewai/cli/train_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def train_crew(n_iterations: int, filename: str) -> None:
     """
@@ -19,7 +21,7 @@ def train_crew(n_iterations: int, filename: str) -> None:
         if not filename.endswith(".pkl"):
             raise ValueError("The filename must not end with .pkl")
 
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

tests/cli/test_subprocess_utils.py

@@ -0,0 +1,140 @@
+import subprocess
+from unittest import mock
+
+import pytest
+
+from crewai.cli.subprocess_utils import run_command
+
+
+class TestRunCommand:
+    """Test the cross-platform subprocess utility."""
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_uses_shell_true(self, mock_subprocess_run, mock_platform):
+        """Test that Windows uses shell=True with proper command conversion."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="uv run test", returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1]["shell"] is True
+        assert isinstance(call_args[0][0], str)
+        assert "uv run test" in call_args[0][0]
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_unix_uses_shell_false(self, mock_subprocess_run, mock_platform):
+        """Test that Unix-like systems use shell=False with list commands."""
+        mock_platform.return_value = "Linux"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["uv", "run", "test"], returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1].get("shell", False) is False
+        assert call_args[0][0] == command
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_command_escaping(self, mock_subprocess_run, mock_platform):
+        """Test that Windows properly escapes command arguments."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="test", returncode=0
+        )
+
+        command = ["echo", "hello world", "test&special"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+        command_str = call_args[0][0]
+
+        assert '"hello world"' in command_str or "'hello world'" in command_str
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_error_handling_preserved(self, mock_subprocess_run, mock_platform):
+        """Test that CalledProcessError is properly raised."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.side_effect = subprocess.CalledProcessError(1, "test")
+
+        with pytest.raises(subprocess.CalledProcessError):
+            run_command(["test"], check=True)
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_all_parameters_passed_through(self, mock_subprocess_run, mock_platform):
+        """Test that all subprocess parameters are properly passed through."""
+        mock_platform.return_value = "Linux"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["test"], returncode=0
+        )
+
+        run_command(
+            ["test"],
+            capture_output=True,
+            text=False,
+            check=False,
+            cwd="/home/test",
+            env={"TEST": "value"},
+            timeout=30
+        )
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1]["capture_output"] is True
+        assert call_args[1]["text"] is False
+        assert call_args[1]["check"] is False
+        assert call_args[1]["cwd"] == "/home/test"
+        assert call_args[1]["env"] == {"TEST": "value"}
+        assert call_args[1]["timeout"] == 30
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_macos_uses_shell_false(self, mock_subprocess_run, mock_platform):
+        """Test that macOS uses shell=False with list commands."""
+        mock_platform.return_value = "Darwin"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["uv", "run", "test"], returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1].get("shell", False) is False
+        assert call_args[0][0] == command
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_string_passthrough(self, mock_subprocess_run, mock_platform):
+        """Test that Windows passes through string commands unchanged."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="test command", returncode=0
+        )
+
+        command_str = "test command with spaces"
+        run_command(command_str)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[0][0] == command_str
+        assert call_args[1]["shell"] is True