Merge branch 'main' into gl/fix/pass-cache-function-to-structured-tool

lib/crewai-tools/src/crewai_tools/tools/file_writer_tool/file_writer_tool.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 from typing import Any
 
 from crewai.tools import BaseTool
@@ -30,27 +31,39 @@ class FileWriterTool(BaseTool):
 
     def _run(self, **kwargs: Any) -> str:
         try:
+            directory = kwargs.get("directory") or "./"
+            filename = kwargs["filename"]
+
+            filepath = os.path.join(directory, filename)
+
+            # Prevent path traversal: the resolved path must be strictly inside
+            # the resolved directory. This blocks ../sequences, absolute paths in
+            # filename, and symlink escapes regardless of how directory is set.
+            # is_relative_to() does a proper path-component comparison that is
+            # safe on case-insensitive filesystems and avoids the "// " edge case
+            # that plagues startswith(real_directory + os.sep).
+            # We also reject the case where filepath resolves to the directory
+            # itself, since that is not a valid file target.
+            real_directory = Path(directory).resolve()
+            real_filepath = Path(filepath).resolve()
+            if not real_filepath.is_relative_to(real_directory) or real_filepath == real_directory:
+                return "Error: Invalid file path — the filename must not escape the target directory."
+
             if kwargs.get("directory"):
-                os.makedirs(kwargs["directory"], exist_ok=True)
+                os.makedirs(real_directory, exist_ok=True)
 
-            # Construct the full path
-            filepath = os.path.join(kwargs.get("directory") or "", kwargs["filename"])
-
-            # Convert overwrite to boolean
             kwargs["overwrite"] = strtobool(kwargs["overwrite"])
 
-            # Check if file exists and overwrite is not allowed
-            if os.path.exists(filepath) and not kwargs["overwrite"]:
-                return f"File {filepath} already exists and overwrite option was not passed."
+            if os.path.exists(real_filepath) and not kwargs["overwrite"]:
+                return f"File {real_filepath} already exists and overwrite option was not passed."
 
-            # Write content to the file
             mode = "w" if kwargs["overwrite"] else "x"
-            with open(filepath, mode) as file:
+            with open(real_filepath, mode) as file:
                 file.write(kwargs["content"])
-            return f"Content successfully written to {filepath}"
+            return f"Content successfully written to {real_filepath}"
         except FileExistsError:
             return (
-                f"File {filepath} already exists and overwrite option was not passed."
+                f"File {real_filepath} already exists and overwrite option was not passed."
             )
         except KeyError as e:
             return f"An error occurred while accessing key: {e!s}"

lib/crewai-tools/tests/tools/test_file_writer_tool.py

@@ -135,3 +135,59 @@ def test_file_exists_error_handling(tool, temp_env, overwrite):
 
     assert "already exists and overwrite option was not passed" in result
     assert read_file(path) == "Pre-existing content"
+
+
+# --- Path traversal prevention ---
+
+def test_blocks_traversal_in_filename(tool, temp_env):
+    # Create a sibling "outside" directory so we can assert nothing was written there.
+    outside_dir = tempfile.mkdtemp()
+    outside_file = os.path.join(outside_dir, "outside.txt")
+    try:
+        result = tool._run(
+            filename=f"../{os.path.basename(outside_dir)}/outside.txt",
+            directory=temp_env["temp_dir"],
+            content="should not be written",
+            overwrite=True,
+        )
+        assert "Error" in result
+        assert not os.path.exists(outside_file)
+    finally:
+        shutil.rmtree(outside_dir, ignore_errors=True)
+
+
+def test_blocks_absolute_path_in_filename(tool, temp_env):
+    # Use a temp file outside temp_dir as the absolute target so we don't
+    # depend on /etc/passwd existing or being writable on the host.
+    outside_dir = tempfile.mkdtemp()
+    outside_file = os.path.join(outside_dir, "target.txt")
+    try:
+        result = tool._run(
+            filename=outside_file,
+            directory=temp_env["temp_dir"],
+            content="should not be written",
+            overwrite=True,
+        )
+        assert "Error" in result
+        assert not os.path.exists(outside_file)
+    finally:
+        shutil.rmtree(outside_dir, ignore_errors=True)
+
+
+def test_blocks_symlink_escape(tool, temp_env):
+    # Symlink inside temp_dir pointing to a separate temp "outside" directory.
+    outside_dir = tempfile.mkdtemp()
+    outside_file = os.path.join(outside_dir, "target.txt")
+    link = os.path.join(temp_env["temp_dir"], "escape")
+    os.symlink(outside_dir, link)
+    try:
+        result = tool._run(
+            filename="escape/target.txt",
+            directory=temp_env["temp_dir"],
+            content="should not be written",
+            overwrite=True,
+        )
+        assert "Error" in result
+        assert not os.path.exists(outside_file)
+    finally:
+        shutil.rmtree(outside_dir, ignore_errors=True)

lib/devtools/src/crewai_devtools/cli.py

@@ -5,6 +5,7 @@ from pathlib import Path
 import subprocess
 import sys
 import time
+from typing import Final, Literal
 
 import click
 from dotenv import load_dotenv
@@ -250,7 +251,9 @@ def add_docs_version(docs_json_path: Path, version: str) -> bool:
     return True
 
 
-_PT_BR_MONTHS = {
+ChangelogLang = Literal["en", "pt-BR", "ko"]
+
+_PT_BR_MONTHS: Final[dict[int, str]] = {
     1: "jan",
     2: "fev",
     3: "mar",
@@ -265,7 +268,9 @@ _PT_BR_MONTHS = {
     12: "dez",
 }
 
-_CHANGELOG_LOCALES: dict[str, dict[str, str]] = {
+_CHANGELOG_LOCALES: Final[
+    dict[ChangelogLang, dict[Literal["link_text", "language_name"], str]]
+] = {
     "en": {
         "link_text": "View release on GitHub",
         "language_name": "English",
@@ -283,7 +288,7 @@ _CHANGELOG_LOCALES: dict[str, dict[str, str]] = {
 
 def translate_release_notes(
     release_notes: str,
-    lang: str,
+    lang: ChangelogLang,
     client: OpenAI,
 ) -> str:
     """Translate release notes into the target language using OpenAI.
@@ -326,7 +331,7 @@ def translate_release_notes(
         return release_notes
 
 
-def _format_changelog_date(lang: str) -> str:
+def _format_changelog_date(lang: ChangelogLang) -> str:
     """Format today's date for a changelog entry in the given language."""
     from datetime import datetime
 
@@ -342,7 +347,7 @@ def update_changelog(
     changelog_path: Path,
     version: str,
     release_notes: str,
-    lang: str = "en",
+    lang: ChangelogLang = "en",
 ) -> bool:
     """Prepend a new release entry to a docs changelog file.
 
@@ -475,6 +480,23 @@ def get_packages(lib_dir: Path) -> list[Path]:
     return packages
 
 
+PrereleaseIndicator = Literal["a", "b", "rc", "alpha", "beta", "dev"]
+_PRERELEASE_INDICATORS: Final[tuple[PrereleaseIndicator, ...]] = (
+    "a",
+    "b",
+    "rc",
+    "alpha",
+    "beta",
+    "dev",
+)
+
+
+def _is_prerelease(version: str) -> bool:
+    """Check if a version string represents a pre-release."""
+    v = version.lower().lstrip("v")
+    return any(indicator in v for indicator in _PRERELEASE_INDICATORS)
+
+
 def get_commits_from_last_tag(tag_name: str, version: str) -> tuple[str, str]:
     """Get commits from the last tag, excluding current version.
 
@@ -489,6 +511,9 @@ def get_commits_from_last_tag(tag_name: str, version: str) -> tuple[str, str]:
         all_tags = run_command(["git", "tag", "--sort=-version:refname"]).split("\n")
         prev_tags = [t for t in all_tags if t and t != tag_name and t != f"v{version}"]
 
+        if not _is_prerelease(version):
+            prev_tags = [t for t in prev_tags if not _is_prerelease(t)]
+
         if prev_tags:
             last_tag = prev_tags[0]
             commit_range = f"{last_tag}..HEAD"
@@ -678,20 +703,28 @@ def _generate_release_notes(
 
     with console.status("[cyan]Generating release notes..."):
         try:
-            prev_bump_commit = run_command(
+            prev_bump_output = run_command(
                 [
                     "git",
                     "log",
                     "--grep=^feat: bump versions to",
-                    "--format=%H",
-                    "-n",
-                    "2",
+                    "--format=%H %s",
                 ]
             )
-            commits_list = prev_bump_commit.strip().split("\n")
+            bump_entries = [
+                line for line in prev_bump_output.strip().split("\n") if line.strip()
+            ]
 
-            if len(commits_list) > 1:
-                prev_commit = commits_list[1]
+            is_stable = not _is_prerelease(version)
+            prev_commit = None
+            for entry in bump_entries[1:]:
+                bump_ver = entry.split("feat: bump versions to", 1)[-1].strip()
+                if is_stable and _is_prerelease(bump_ver):
+                    continue
+                prev_commit = entry.split()[0]
+                break
+
+            if prev_commit:
                 commit_range = f"{prev_commit}..HEAD"
                 commits = run_command(
                     ["git", "log", commit_range, "--pretty=format:%s"]
@@ -777,10 +810,7 @@ def _generate_release_notes(
             "\n[green]✓[/green] Using generated release notes without editing"
         )
 
-    is_prerelease = any(
-        indicator in version.lower()
-        for indicator in ["a", "b", "rc", "alpha", "beta", "dev"]
-    )
+    is_prerelease = _is_prerelease(version)
 
     return release_notes, openai_client, is_prerelease
 
@@ -799,7 +829,7 @@ def _update_docs_and_create_pr(
         The docs branch name if a PR was created, None otherwise.
     """
     docs_json_path = cwd / "docs" / "docs.json"
-    changelog_langs = ["en", "pt-BR", "ko"]
+    changelog_langs: list[ChangelogLang] = ["en", "pt-BR", "ko"]
 
     if not dry_run:
         docs_files_staged: list[str] = []