Skip to content
Merged
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
2 changes: 1 addition & 1 deletion .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
43 changes: 37 additions & 6 deletions cloud_storage/cloud_storage/overrides/file.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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:
"""
Expand Down Expand Up @@ -348,14 +377,15 @@ 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
"""
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
Expand All @@ -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:
Expand Down
20 changes: 20 additions & 0 deletions cloud_storage/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down
2 changes: 1 addition & 1 deletion cloud_storage/tests/setup.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down
19 changes: 10 additions & 9 deletions cloud_storage/tests/test_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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}"
96 changes: 96 additions & 0 deletions cloud_storage/tests/test_local_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -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"
Expand Down
Loading