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
5 changes: 5 additions & 0 deletions apps/backend/alembic.ini
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
[alembic]
script_location = alembic
prepend_sys_path = .
# Alembic 1.16+ warns when this is unset and falls back to legacy splitting on
# spaces, commas and colons. `os` uses the platform path separator, which is
# the correct reading of the single "." entry above and keeps the behaviour
# stable when the legacy fallback is eventually removed.
path_separator = os
sqlalchemy.url = sqlite:///./.local/partha.db

[loggers]
Expand Down
12 changes: 11 additions & 1 deletion apps/backend/app/storage/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,17 @@ def _safe_extract_tar(self, archive: tarfile.TarFile, destination: Path) -> None
"Archive would decompress to more than the configured maximum size.",
{"maxExtractedSizeBytes": self.max_extracted_size_bytes},
)
archive.extractall(destination)
# `filter="data"` applies CPython's own extraction hardening: it strips
# absolute paths and `..` traversal, and rejects links, devices, setuid
# bits and other unsafe metadata as the members are written.
#
# The loop above already rejects those cases, so this is defence in
# depth on untrusted uploads rather than the primary control — the two
# have to disagree for it to matter, which is exactly when a check is
# worth having. It also settles the DeprecationWarning: tar extraction
# is unfiltered by default until Python 3.14, which would switch this
# behaviour on silently. Being explicit keeps it a decision.
archive.extractall(destination, filter="data")

def _normalise_single_root(self, destination: Path) -> Path:
children = [child for child in destination.iterdir() if child.name != "__MACOSX"]
Expand Down
67 changes: 67 additions & 0 deletions apps/backend/tests/test_ingestion_resource_budgets.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,73 @@ def fake_clone(_: GitHubClient, __: str, destination: Path, ___: str | None = No
assert _repositories_dir_is_empty(file_count_limited_client.storage_path) # type: ignore[attr-defined]


# --- archive path safety ------------------------------------------------------
#
# These cover the guards in ``LocalStorage._safe_extract_tar`` /
# ``_safe_extract_zip`` that reject traversal and link members. They were
# previously untested, so a regression would have been silent — and the tar
# path additionally relies on ``extractall(filter="data")`` as defence in depth.


def _malicious_tar_gz_bytes(members: list[tarfile.TarInfo], payload: bytes = b"pwned") -> bytes:
"""A tar built from raw ``TarInfo`` objects, so unsafe members can be forged."""
buffer = io.BytesIO()
with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
for info in members:
if info.isreg():
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))
else:
archive.addfile(info)
return buffer.getvalue()


def test_tar_archive_with_parent_traversal_path_is_rejected(auth_client):
"""A member escaping the destination via ``..`` must never be written."""
escaping = tarfile.TarInfo("../escaped.txt")
escaping.type = tarfile.REGTYPE

response = _upload(auth_client, "evil.tar.gz", _malicious_tar_gz_bytes([escaping]))

error = assert_error_response(response, 422, "validation_error")
assert error.message == "Archive contains unsafe paths."


def test_tar_archive_with_absolute_path_is_rejected(auth_client):
"""An absolute member path must not be able to write outside the sandbox."""
absolute = tarfile.TarInfo("/tmp/partha-escaped.txt")
absolute.type = tarfile.REGTYPE

response = _upload(auth_client, "evil.tar.gz", _malicious_tar_gz_bytes([absolute]))

error = assert_error_response(response, 422, "validation_error")
assert error.message == "Archive contains unsafe paths."


def test_tar_archive_with_symlink_member_is_rejected(auth_client):
"""Symlinks are refused outright: they are the classic extraction escape."""
link = tarfile.TarInfo("sample/link")
link.type = tarfile.SYMTYPE
link.linkname = "/etc/passwd"

response = _upload(auth_client, "evil.tar.gz", _malicious_tar_gz_bytes([link]))

error = assert_error_response(response, 422, "validation_error")
assert error.message == "Archive contains unsupported link or device entries."


def test_zip_archive_with_parent_traversal_path_is_rejected(auth_client):
"""The zip path enforces the same containment rule as the tar path."""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr("../escaped.txt", "pwned")

response = _upload(auth_client, "evil.zip", buffer.getvalue())

error = assert_error_response(response, 422, "validation_error")
assert error.message == "Archive contains unsafe paths."


# --- regression: unaffected happy path --------------------------------------


Expand Down
Loading