Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -402,7 +402,41 @@ def set_feature(item, feature_path: _VisitPath):
elif len(feature_path) == 0:
if item is not None:
file_relpath = os.path.normpath(item).replace("\\", "/")
# Guard against path traversal (CWE-22): a crafted `file_name` such as
# "../../etc/passwd" or an absolute path must not be able to escape the
# metadata file's directory and read arbitrary files on the host. The
# check is performed on the relative reference (not the joined result)
# so that it works both for local directories and for the fsspec URLs
# (e.g. "zip://...::...") used when reading from downloaded archives.
if (
os.path.isabs(item)
or os.path.isabs(file_relpath)
or file_relpath == ".."
or file_relpath.startswith("../")
):
raise ValueError(
f"Invalid metadata file_name '{item}': `file_name` must be a relative path "
f"pointing inside the directory containing the metadata file. Absolute paths "
f"and parent-directory ('..') traversal that escape the dataset directory are "
f"not allowed."
)
item = os.path.join(downloaded_metadata_dir, file_relpath)
# For local paths, additionally resolve symlinks and confirm containment.
# Skipped for fsspec URLs (which contain "://") since realpath is not
# aware of them and would corrupt the URL.
if "://" not in item:
Comment on lines +424 to +427

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you can point to local files using file://../.. or local://../.. so this check is maybe not enough

maybe we need to forbid :// altogether ? (except for zip://<file_name>::<valid_relative_path>)

real_root = os.path.realpath(downloaded_metadata_dir)
real_path = os.path.realpath(item)
try:
is_contained = os.path.commonpath([real_root, real_path]) == real_root
except ValueError:
# commonpath raises ValueError for paths on different drives (Windows)
is_contained = False
if not is_contained:
raise ValueError(
f"Invalid metadata file_name '{item}': the resolved path escapes the "
f"dataset directory containing the metadata file."
)
return item

for pa_metadata_table in self._read_metadata(downloaded_metadata_file, metadata_ext=metadata_ext):
Expand Down
84 changes: 84 additions & 0 deletions tests/packaged_modules/test_folder_based_builder.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import importlib
import json
import os
import shutil
import textwrap

Expand Down Expand Up @@ -449,6 +451,88 @@ def test_data_files_with_wrong_metadata_file_name(cache_dir, tmp_path, auto_text
assert all("additional_feature" not in example for _, example in generator)


@pytest.mark.parametrize(
"malicious_file_name",
[
"../outside/secret.txt", # relative traversal
"../../outside/secret.txt", # deeper relative traversal
"subdir/../../outside/secret.txt", # traversal hidden after a valid segment
],
)
def test_data_files_with_metadata_path_traversal_is_rejected(cache_dir, tmp_path, auto_text_file, malicious_file_name):
# Regression test for https://github.com/huggingface/datasets/issues/8324
# A crafted `file_name` must not be able to escape the metadata file's directory.
secret_dir = tmp_path / "outside"
secret_dir.mkdir(parents=True, exist_ok=True)
shutil.copyfile(auto_text_file, secret_dir / "secret.txt")

data_dir = tmp_path / "data_dir_with_traversal_metadata"
data_dir.mkdir(parents=True, exist_ok=True)
shutil.copyfile(auto_text_file, data_dir / "file.txt")
metadata_filename = data_dir / "metadata.jsonl"
metadata = f'{{"file_name": {json.dumps(malicious_file_name)}, "additional_feature": "Malicious file"}}\n'
with open(metadata_filename, "w", encoding="utf-8") as f:
f.write(metadata)

data_files = DataFilesDict.from_patterns(get_data_patterns(str(data_dir)), data_dir.as_posix())
autofolder = DummyFolderBasedBuilder(data_files=data_files, cache_dir=cache_dir)
gen_kwargs = autofolder._split_generators(StreamingDownloadManager())[0].gen_kwargs
with pytest.raises(ValueError, match="Invalid metadata file_name"):
list(autofolder._generate_examples(**gen_kwargs))


def test_data_files_with_metadata_absolute_path_is_rejected(cache_dir, tmp_path, auto_text_file):
# Regression test for https://github.com/huggingface/datasets/issues/8324
# An absolute `file_name` must not be able to read files outside the dataset directory.
secret_dir = tmp_path / "outside"
secret_dir.mkdir(parents=True, exist_ok=True)
secret_file = secret_dir / "secret.txt"
shutil.copyfile(auto_text_file, secret_file)

data_dir = tmp_path / "data_dir_with_absolute_metadata"
data_dir.mkdir(parents=True, exist_ok=True)
shutil.copyfile(auto_text_file, data_dir / "file.txt")
metadata_filename = data_dir / "metadata.jsonl"
metadata = f'{{"file_name": {json.dumps(str(secret_file))}, "additional_feature": "Malicious file"}}\n'
with open(metadata_filename, "w", encoding="utf-8") as f:
f.write(metadata)

data_files = DataFilesDict.from_patterns(get_data_patterns(str(data_dir)), data_dir.as_posix())
autofolder = DummyFolderBasedBuilder(data_files=data_files, cache_dir=cache_dir)
gen_kwargs = autofolder._split_generators(StreamingDownloadManager())[0].gen_kwargs
with pytest.raises(ValueError, match="Invalid metadata file_name"):
list(autofolder._generate_examples(**gen_kwargs))


def test_data_files_with_metadata_legitimate_subdir_reference(cache_dir, tmp_path, auto_text_file):
# A legitimate `file_name` pointing to a file in a subdirectory of the metadata
# file's directory must keep working after the path traversal fix (#8324).
data_dir = tmp_path / "data_dir_with_subdir_metadata"
data_dir.mkdir(parents=True, exist_ok=True)
subdir = data_dir / "subdir"
subdir.mkdir(parents=True, exist_ok=True)
shutil.copyfile(auto_text_file, data_dir / "file.txt")
shutil.copyfile(auto_text_file, subdir / "file2.txt")
metadata_filename = data_dir / "metadata.jsonl"
metadata = textwrap.dedent(
"""\
{"file_name": "file.txt", "additional_feature": "Same-dir file"}
{"file_name": "subdir/file2.txt", "additional_feature": "Subdir file"}
{"file_name": "./subdir/file2.txt", "additional_feature": "Subdir file with dot prefix"}
"""
)
with open(metadata_filename, "w", encoding="utf-8") as f:
f.write(metadata)

data_files = DataFilesDict.from_patterns(get_data_patterns(str(data_dir)), data_dir.as_posix())
autofolder = DummyFolderBasedBuilder(data_files=data_files, cache_dir=cache_dir)
gen_kwargs = autofolder._split_generators(StreamingDownloadManager())[0].gen_kwargs
examples = [example for _, example in autofolder._generate_examples(**gen_kwargs)]
assert len(examples) == 3
assert all(example["base"] is not None for example in examples)
assert all(os.path.isfile(example["base"]) for example in examples)


def test_data_files_with_custom_file_name_column_in_metadata_file(cache_dir, tmp_path, auto_text_file):
data_dir = tmp_path / "data_dir_with_custom_file_name_metadata"
data_dir.mkdir(parents=True, exist_ok=True)
Expand Down