diff --git a/lib/crewai-tools/src/crewai_tools/tools/files_compressor_tool/files_compressor_tool.py b/lib/crewai-tools/src/crewai_tools/tools/files_compressor_tool/files_compressor_tool.py index 5955ea523..603486ba6 100644 --- a/lib/crewai-tools/src/crewai_tools/tools/files_compressor_tool/files_compressor_tool.py +++ b/lib/crewai-tools/src/crewai_tools/tools/files_compressor_tool/files_compressor_tool.py @@ -128,6 +128,7 @@ class FileCompressorTool(BaseTool): # Defense in depth: validate the write target at the sink, so this is # safe even if called directly rather than through _run. output_path = validate_file_path(output_path) + input_path = validate_file_path(input_path) skipped: list[str] = [] with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zipf: if os.path.isfile(input_path): @@ -137,12 +138,12 @@ class FileCompressorTool(BaseTool): for file in files: full_path = os.path.join(root, file) try: - validate_file_path(full_path) + resolved_path = validate_file_path(full_path) except ValueError: skipped.append(full_path) continue arcname = os.path.relpath(full_path, start=input_path) - zipf.write(full_path, arcname) + zipf.write(resolved_path, arcname) return skipped @staticmethod @@ -177,6 +178,7 @@ class FileCompressorTool(BaseTool): # Defense in depth: validate the write target at the sink, so this is # safe even if called directly rather than through _run. output_path = validate_file_path(output_path) + input_path = validate_file_path(input_path) with tarfile.open(output_path, mode) as tarf: # type: ignore[call-overload] arcname = os.path.basename(input_path) tarf.add(input_path, arcname=arcname, filter=_drop_links) diff --git a/lib/crewai-tools/tests/tools/files_compressor_tool_test.py b/lib/crewai-tools/tests/tools/files_compressor_tool_test.py index 668ca9780..70bd9e02a 100644 --- a/lib/crewai-tools/tests/tools/files_compressor_tool_test.py +++ b/lib/crewai-tools/tests/tools/files_compressor_tool_test.py @@ -165,7 +165,12 @@ def symlink_env(): prev_cwd = os.getcwd() os.chdir(work_dir) - yield {"work_dir": work_dir, "src": src, "secret_dir": secret_dir} + yield { + "work_dir": work_dir, + "src": src, + "secret_dir": secret_dir, + "secret_file": secret_file, + } os.chdir(prev_cwd) shutil.rmtree(work_dir, ignore_errors=True) shutil.rmtree(secret_dir, ignore_errors=True) @@ -219,3 +224,17 @@ def test_compress_tar_validates_output_path_at_sink(symlink_env): with pytest.raises(ValueError, match="outside the allowed director"): FileCompressorTool._compress_tar(symlink_env["src"], outside, "tar.gz") assert not os.path.exists(outside) + + +def test_compress_zip_validates_input_path_at_sink(symlink_env): + out = os.path.join(symlink_env["work_dir"], "archive.zip") + with pytest.raises(ValueError, match="outside the allowed director"): + FileCompressorTool._compress_zip(symlink_env["secret_file"], out) + assert not os.path.exists(out) + + +def test_compress_tar_validates_input_path_at_sink(symlink_env): + out = os.path.join(symlink_env["work_dir"], "archive.tar.gz") + with pytest.raises(ValueError, match="outside the allowed director"): + FileCompressorTool._compress_tar(symlink_env["secret_file"], out, "tar.gz") + assert not os.path.exists(out)