andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-06 15:39:24 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

fix: validate compressor input paths at archive sinks

Browse Source
This commit is contained in:
Rip&Tear
2026-06-26 09:15:35 +08:00
parent 034b119d34
commit e584b0135d
2 changed files with 24 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
lib/crewai-tools/src/crewai_tools/tools/files_compressor_tool/files_compressor_tool.py
View File

@@ -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)

21
lib/crewai-tools/tests/tools/files_compressor_tool_test.py
View File

@@ -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)