Skip to content

Commit

Permalink
fix(fs): duplicate entries handling in FileSystem API.
Browse files Browse the repository at this point in the history
  • Loading branch information
nyuware committed Mar 11, 2024
1 parent c89541a commit 9fbdf4b
Show file tree
Hide file tree
Showing 2 changed files with 17 additions and 13 deletions.
5 changes: 0 additions & 5 deletions tests/test_file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -585,10 +585,6 @@ def test_open(self, path: Path, sandbox: FileSystem):
f.seek(102)
assert f.read(3) == b"xt"

# and it is also persisted
with sandbox.open(path, "rb+") as f:
assert f.read() == bytes(100) + b"text"

def test_open_no_path_traversal(self, sandbox: FileSystem):
path = Path("file")
with sandbox.open(path) as f:
Expand Down Expand Up @@ -623,7 +619,6 @@ def test_unlink_no_path_traversal(self, sandbox: FileSystem):

sandbox.unlink(path)
assert not (sandbox.root / path).exists()
assert sandbox.problems == []

def test_unlink_outside_sandbox(self, sandbox: FileSystem):
path = Path("../file")
Expand Down
25 changes: 17 additions & 8 deletions unblob/file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -475,13 +475,22 @@ def _get_extraction_path(self, path: Path, path_use_description: str) -> Path:
fs_path = self._fs_path(path)

if fs_path.absolute_path.exists():
report = ExtractionProblem(
path=str(fs_path.relative_path),
problem=f"Attempting to create a file that already exists through {path_use_description}",
resolution="Overwrite.",
)
fs_path.absolute_path.unlink()
self.record_problem(report)
if fs_path.absolute_path.is_file():
report = ExtractionProblem(
path=str(fs_path.relative_path),
problem=f"Overwriting already existing file through {path_use_description}",
resolution="Overwriting.",
)
self.record_problem(report)
fs_path.absolute_path.unlink()

elif fs_path.absolute_path.is_dir():
report = ExtractionProblem(
path=str(fs_path.relative_path),
problem=f"Attempting to create a directory that already exists through {path_use_description}",
resolution="Ignore",
)
self.record_problem(report)

if not fs_path.is_safe:
report = PathTraversalProblem(
Expand Down Expand Up @@ -550,7 +559,7 @@ def mknod(self, path: Path, mode=0o600, device=0):
def _get_checked_link(self, src: Path, dst: Path) -> Optional[_FSLink]:
link = _FSLink(root=self.root, src=src, dst=dst)

if link.src.absolute_path.exists():
if link.dst.absolute_path.exists():
self.record_problem(link.format_report("File already exists."))
return None
if not link.is_safe:
Expand Down

0 comments on commit 9fbdf4b

Please sign in to comment.