fix: re-validate interpolated output_file against untrusted inputs

.github/workflows/vulnerability-scan.yml

@@ -51,6 +51,7 @@ jobs:
             --ignore-vuln PYSEC-2024-277 \
             --ignore-vuln PYSEC-2026-89 \
             --ignore-vuln PYSEC-2026-97 \
+            --ignore-vuln PYSEC-2026-597 \
             --ignore-vuln PYSEC-2025-148 \
             --ignore-vuln PYSEC-2025-183 \
             --ignore-vuln PYSEC-2025-189 \
@@ -78,6 +79,7 @@ jobs:
         #   PYSEC-2024-277      - joblib 1.5.3: disputed; NumpyArrayWrapper only used with trusted caches
         #   PYSEC-2026-89       - markdown 3.10.2: DoS via malformed HTML; fix 3.8.1 — already past, advisory range is stale
         #   PYSEC-2026-97       - nltk 3.9.4: arbitrary file read in filestring(); no fix available
+        #   PYSEC-2026-597      - nltk 3.9.4: path traversal via percent-encoded sequences in data.load()/find() (incomplete fix for #3504); latest release, no fix available. Transitive via unstructured; crewai never calls nltk.data.load/find with untrusted input.
         #   PYSEC-2025-148      - onnx 1.21.0: path traversal in save_external_data; no fix available
         #   PYSEC-2025-183      - pyjwt 2.12.1: disputed weak-encryption claim; key length is application-chosen
         #   PYSEC-2025-189..197 - torch 2.11.0: memory-corruption/DoS in functions only reachable via untrusted models; no fix available

lib/crewai/src/crewai/task.py

@@ -10,7 +10,7 @@
 import inspect
 import json
 import logging
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 import threading
 from typing import (
     Annotated,
@@ -504,6 +504,29 @@ def output_file_validation(cls, value: str | None) -> str | None:
         if value is None:
             return None
 
+        if "{" in value or "}" in value:
+            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
+            for var in template_vars:
+                if not var.isidentifier():
+                    raise ValueError(f"Invalid template variable name: {var}")
+            # Literal portions are still checked here; the fully interpolated
+            # path is re-validated at runtime (see
+            # interpolate_inputs_and_add_conversation_history) because template
+            # variables may be filled from untrusted kickoff inputs.
+            cls._sanitize_output_file_path(value)
+            return value
+
+        return cls._sanitize_output_file_path(value)
+
+    @staticmethod
+    def _sanitize_output_file_path(value: str) -> str:
+        """Enforce path-safety on an ``output_file`` value.
+
+        Shared by the field validator's literal and template branches. Rejects
+        traversal sequences, shell expansion, and shell metacharacters, and
+        strips a leading ``/`` so a literal path stays relative to the working
+        directory.
+        """
         if ".." in value:
             raise ValueError(
                 "Path traversal attempts are not allowed in output_file paths"
@@ -519,17 +542,56 @@ def output_file_validation(cls, value: str | None) -> str | None:
                 "Shell special characters are not allowed in output_file paths"
             )
 
-        if "{" in value or "}" in value:
-            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
-            for var in template_vars:
-                if not var.isidentifier():
-                    raise ValueError(f"Invalid template variable name: {var}")
-            return value
-
         if value.startswith("/"):
             return value[1:]
         return value
 
+    def _validate_output_file_input_values(
+        self, inputs: dict[str, str | int | float | dict[str, Any] | list[Any]]
+    ) -> None:
+        """Reject untrusted input values that would escape the output path.
+
+        Only the variables that actually appear in the ``output_file`` template
+        are checked. The developer-authored template is trusted (it may contain
+        an absolute base directory), but a value substituted into it must not
+        introduce path traversal (``..``), an absolute path, a home/variable
+        expansion (``~``/``$``), or shell metacharacters that would redirect the
+        write outside the intended location.
+        """
+        if not self._original_output_file:
+            return
+
+        template_vars = [
+            part.split("}")[0] for part in self._original_output_file.split("{")[1:]
+        ]
+        for var in template_vars:
+            if var not in inputs:
+                continue
+            value = str(inputs[var])
+            if ".." in value:
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': path "
+                    "traversal sequences ('..') are not allowed"
+                )
+            if value.startswith(("~", "$")):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': shell "
+                    "expansion characters are not allowed"
+                )
+            if any(char in value for char in ["|", ">", "<", "&", ";"]):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': shell "
+                    "special characters are not allowed"
+                )
+            if (
+                PurePosixPath(value).is_absolute()
+                or PureWindowsPath(value).is_absolute()
+            ):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': absolute "
+                    "paths are not allowed"
+                )
+
     @model_validator(mode="after")
     def set_attributes_based_on_config(self) -> Task:
         """Set attributes based on the agent configuration."""
@@ -1026,6 +1088,12 @@ def interpolate_inputs_and_add_conversation_history(
             raise ValueError(f"Error interpolating expected_output: {e!s}") from e
 
         if self.output_file is not None:
+            # Values interpolated into the output path may come from untrusted
+            # kickoff inputs. The developer-authored template (including any
+            # absolute base directory) is trusted, but an injected value must
+            # not introduce path traversal, an absolute path, or shell
+            # expansion that would escape the intended location.
+            self._validate_output_file_input_values(inputs)
             try:
                 self.output_file = interpolate_only(
                     input_string=self._original_output_file, inputs=inputs

lib/crewai/tests/test_task.py

@@ -931,6 +931,48 @@ def test_interpolate_inputs(tmp_path):
     assert task.output_file == str(tmp_path / "ML" / "output_2025.txt")
 
 
+@pytest.mark.parametrize(
+    ("template", "malicious_inputs", "expected_error"),
+    [
+        ("reports/{name}.md", {"name": "../../../../tmp/pwn"}, "path traversal"),
+        ("{p}", {"p": "/tmp/abs_pwn"}, "absolute paths"),
+        ("{p}", {"p": "~/.bashrc"}, "shell expansion"),
+        ("{p}", {"p": "x;rm -rf /"}, "shell special characters"),
+        ("{p}", {"p": r"C:\Windows\evil"}, "absolute paths"),
+    ],
+)
+def test_interpolate_output_file_rejects_unsafe_inputs(
+    template, malicious_inputs, expected_error
+):
+    """Untrusted inputs must not escape the output_file path via interpolation."""
+    task = Task(
+        description="do {p} {name}".replace("{p}", "x").replace("{name}", "x"),
+        expected_output="e",
+        output_file=template,
+    )
+    with pytest.raises(ValueError, match=expected_error):
+        task.interpolate_inputs_and_add_conversation_history(inputs=malicious_inputs)
+
+
+def test_interpolate_output_file_allows_safe_inputs(tmp_path):
+    """Safe input values and developer-chosen absolute base paths still work."""
+    task = Task(
+        description="d",
+        expected_output="e",
+        output_file="reports/{name}.md",
+    )
+    task.interpolate_inputs_and_add_conversation_history(inputs={"name": "q3_summary"})
+    assert task.output_file == "reports/q3_summary.md"
+
+    abs_task = Task(
+        description="d",
+        expected_output="e",
+        output_file=str(tmp_path / "{topic}" / "out.md"),
+    )
+    abs_task.interpolate_inputs_and_add_conversation_history(inputs={"topic": "sales"})
+    assert abs_task.output_file == str(tmp_path / "sales" / "out.md")
+
+
 def test_interpolate_only():
     """Test the interpolate_only method for various scenarios including JSON structure preservation."""