diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6af1a95..818215f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -45,7 +45,7 @@ repos: additional_dependencies: ['flake8-bugbear'] - repo: https://github.com/agritheory/test_utils - rev: v1.28.0 + rev: v1.30.1 hooks: - id: update_pre_commit_config - id: validate_frappe_project diff --git a/cloud_storage/cloud_storage/overrides/file.py b/cloud_storage/cloud_storage/overrides/file.py index ad7523a..0e8808c 100644 --- a/cloud_storage/cloud_storage/overrides/file.py +++ b/cloud_storage/cloud_storage/overrides/file.py @@ -68,13 +68,34 @@ def unique_url(self) -> str: separator = "&" if "?" in file_url else "?" return file_url + separator + urlencode({"fid": self.name}) + def validate_file_path(self, path=None): + """ + HASH: 48366c6ecbad44ed24e6d02bdd8f8f189ce58927 + REPO: https://github.com/frappe/frappe + PATH: frappe/core/doctype/file/file.py + METHOD: validate_file_path + """ + if path is None: + if self.is_remote_file: + return + path = self.get_full_path() + base_path = os.path.realpath(get_files_path(is_private=self.is_private)) + resolved_path = os.path.realpath(path) + if os.path.commonpath((base_path, resolved_path)) != base_path: + frappe.throw(_("The File URL you've entered is incorrect"), title=_("Invalid File URL")) + def validate(self) -> None: """ - HASH: 69a495579a729909f4df7a45855165eee4a208f4 + HASH: ca4a6912ebf7a8dafa9e56822bd304fa38c4da09 REPO: https://github.com/frappe/frappe PATH: frappe/core/doctype/file/file.py METHOD: validate """ + if self.is_folder: + if self.file_url: + frappe.throw(_("A folder cannot have a File URL")) + return + # guard against recursion: associate_files() can save another File, re-entering validate if not self.flags.associating_files: self.associate_files() @@ -211,14 +232,22 @@ def after_insert(self) -> File: self.point_at_surviving_file(associated_doc) def point_at_surviving_file(self, surviving_name: str) -> None: - """Repoint this document at the file it was just merged into + """Repoint this document at the file it was just merged into. """ surviving = frappe.db.get_value( "File", surviving_name, ["file_url", "s3_key"], as_dict=True ) - if surviving: - self.file_url = surviving.file_url - self.s3_key = surviving.s3_key + if not surviving: + return + + merged_name = (self.doctype, self.name) + self.name = surviving_name + self.file_url = surviving.file_url + self.s3_key = surviving.s3_key + + if merged_name in frappe.flags.currently_saving: + frappe.flags.currently_saving.remove(merged_name) + frappe.flags.currently_saving.append((self.doctype, self.name)) def on_trash(self) -> None: """ @@ -348,7 +377,7 @@ def remove_file_association(self, dt: str, dn: str) -> None: @frappe.whitelist() def get_content(self) -> bytes: """ - HASH: bfbebb3d3d9c26eb34ed447112fcd46f1dadff00 + HASH: 48366c6ecbad44ed24e6d02bdd8f8f189ce58927 REPO: https://github.com/frappe/frappe PATH: frappe/core/doctype/file/file.py METHOD: get_content @@ -356,6 +385,7 @@ def get_content(self) -> bytes: if self.is_folder: frappe.throw(_("Cannot get file contents of a Folder")) + self.validate_file_path() if self.get("content"): self._content = self.content if self.decode: # type: ignore @@ -378,6 +408,7 @@ def get_content(self) -> bytes: file_path = frappe.get_site_path("public", "files", self.file_name) else: file_path = frappe.get_site_path("private", "files", self.file_name) + self.validate_file_path(file_path) with open(file_path, mode="rb") as f: self._content = f.read() try: diff --git a/cloud_storage/tests/conftest.py b/cloud_storage/tests/conftest.py index ae770ff..b772cc8 100644 --- a/cloud_storage/tests/conftest.py +++ b/cloud_storage/tests/conftest.py @@ -82,6 +82,26 @@ def patch_frappe_conf(monkeymodule): ) +@pytest.fixture(scope="session", autouse=True) +def cleanup_uploaded_files(db_instance): + watched_dirs = [ + Path(frappe.get_site_path("public", "files")), + Path(frappe.get_site_path("private", "files")), + ] + before = {d: (set(d.rglob("*")) if d.is_dir() else set()) for d in watched_dirs} + + yield + + for d in watched_dirs: + after = set(d.rglob("*")) if d.is_dir() else set() + new_paths = sorted(after - before[d], key=lambda p: len(p.parts), reverse=True) + for path in new_paths: + if path.is_file(): + path.unlink(missing_ok=True) + elif path.is_dir() and not any(path.iterdir()): + path.rmdir() + + @pytest.fixture def mocked_s3_client(): with mock_s3(): diff --git a/cloud_storage/tests/setup.py b/cloud_storage/tests/setup.py index aff7d36..293f9d0 100644 --- a/cloud_storage/tests/setup.py +++ b/cloud_storage/tests/setup.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025, AgriTheory and contributors +# Copyright (c) 2026, AgriTheory and contributors # For license information, please see license.txt import frappe diff --git a/cloud_storage/tests/test_file.py b/cloud_storage/tests/test_file.py index 6da3142..62264af 100644 --- a/cloud_storage/tests/test_file.py +++ b/cloud_storage/tests/test_file.py @@ -160,7 +160,7 @@ def test_upload_file_with_multiple_association(example_file_record_1): _file.load_from_db() assert frappe.db.exists("File", _file.name) is not None - assert frappe.db.exists("File", file.name) is None + assert file.name == _file.name assert len(_file.file_association) >= 2 assert _file.file_association[0].link_doctype == "User" assert _file.file_association[0].link_name == "Administrator" @@ -228,7 +228,7 @@ def test_file_versioning_with_content_change(mocked_s3_client, example_file_reco file2 = create_upload_file(modified_csv, file_name="sample.csv") file1.load_from_db() - assert not frappe.db.exists("File", file2.name) + assert file2.name == file1.name assert file1.content_hash == file2.content_hash assert file1.content_hash != original_content_hash assert len(file1.versions) >= 2 @@ -255,8 +255,8 @@ def test_file_versioning_different_documents(mocked_s3_client, example_file_reco ) file1.load_from_db() - # No duplicate record should exist - assert not frappe.db.exists("File", file2.name) + # No duplicate record should exist: the second upload points at the first + assert file2.name == file1.name # Existing record must keep its s3_key assert file1.s3_key is not None # Both documents should be associated @@ -337,7 +337,6 @@ def test_associate_files_no_duplicate_association(): doc.associate_files("Module Def", "Cloud Storage") doc.associate_files("Module Def", "Cloud Storage") - file_b = frappe.get_doc("File", "test-assoc-guard-b") matching = [ a @@ -487,11 +486,13 @@ def paste(file_name, attached_to_name): second = paste("dup-paste-second.png", "Guest") - # the duplicate is merged into the existing file and deleted during its own insert - assert not frappe.db.exists("File", second.name) + # the duplicate is merged into the existing file and deleted during its own insert, + # so the document has to come back pointing at the survivor + assert second.name == first.name + assert frappe.db.exists("File", second.name) - # ...but the URL it hands back must still resolve + # ...and the URL it hands back must still resolve assert "?key=" in second.file_url key = second.file_url.split("?key=")[1] assert frappe.db.exists("File", {"s3_key": key}) - assert second.unique_url == f"{second.file_url}&fid={second.name}" + assert second.unique_url == f"{second.file_url}&fid={first.name}" diff --git a/cloud_storage/tests/test_local_storage.py b/cloud_storage/tests/test_local_storage.py index 0042ad5..acf2d1e 100644 --- a/cloud_storage/tests/test_local_storage.py +++ b/cloud_storage/tests/test_local_storage.py @@ -120,3 +120,99 @@ def test_delete_file_local(example_file): file.delete() assert not frappe.db.exists("File", file.name) + + +def test_get_content_rejects_absolute_path_escape(): + """An absolute file_name escapes the files/ sandbox via os.path.join.""" + frappe.set_user("Administrator") + file = frappe.new_doc("File") + file.file_name = "/etc/hostname" + file.file_url = "" + file.is_private = 0 + + with pytest.raises(frappe.ValidationError): + file.get_content() + + +def test_get_content_rejects_relative_path_escape(): + """Same escape, via ../ traversal instead of an absolute path.""" + frappe.set_user("Administrator") + file = frappe.new_doc("File") + file.file_name = "../" * 10 + "etc/hostname" + file.file_url = "" + file.is_private = 1 + + with pytest.raises(frappe.ValidationError): + file.get_content() + + +def test_get_content_rejects_sibling_directory_escape(): + """A sibling dir that merely shares the files/ prefix as a string (files-shadow) + must not pass containment just because startswith(base_path) is true.""" + frappe.set_user("Administrator") + shadow_dir = Path(frappe.get_site_path("private")) / "files-shadow" + shadow_dir.mkdir(exist_ok=True) + secret = shadow_dir / "secret.txt" + secret.write_text("outside the sandbox") + + file = frappe.new_doc("File") + file.file_name = "../files-shadow/secret.txt" + file.file_url = "" + file.is_private = 1 + + try: + with pytest.raises(frappe.ValidationError): + file.get_content() + finally: + secret.unlink() + shadow_dir.rmdir() + + +def test_duplicate_content_unique_url_resolves_locally(): + """Pasting an image whose content already exists must yield a fid that still exists. + + Without cloud storage, /private/files URLs are resolved by + frappe.utils.response.download_private_file through + find_file_by_url(path, name=fid), so the fid in unique_url has to name the + File the duplicate was merged into, not the deleted duplicate itself. + """ + from urllib.parse import parse_qs, urlsplit + + from frappe.core.doctype.file.utils import find_file_by_url + from PIL import Image + + frappe.set_user("Administrator") + + buffer = BytesIO() + Image.new("RGB", (5, 3), (211, 47, 12)).save(buffer, format="PNG") + content = buffer.getvalue() + + def paste(file_name, attached_to_name): + doc = frappe.get_doc( + { + "doctype": "File", + "file_name": file_name, + "attached_to_doctype": "User", + "attached_to_name": attached_to_name, + "content": content, + "decode": False, + "is_private": 1, + } + ) + doc.save(ignore_permissions=True) + return doc + + first = paste("dup-local-first.png", "Administrator") + assert frappe.db.exists("File", first.name) + assert first.file_url.startswith("/private/files/") + + second = paste("dup-local-second.png", "Guest") + + assert second.name == first.name + assert second.file_url == first.file_url + assert second.unique_url == f"{first.file_url}?fid={first.name}" + + # resolve it the way download_private_file does + url = urlsplit(second.unique_url) + fid = parse_qs(url.query)["fid"][0] + assert find_file_by_url(url.path, name=fid) is not None diff --git a/pyproject.toml b/pyproject.toml index 4fc31fc..9fa0bf5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cloud_storage" -version = "15.10.3" +version = "15.10.5" description = "Frappe App for integrating with cloud storage applications" readme = "README.md" requires-python = ">=3.10"