diff --git a/apps/backend/alembic.ini b/apps/backend/alembic.ini index 87070c5..ebf9135 100644 --- a/apps/backend/alembic.ini +++ b/apps/backend/alembic.ini @@ -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] diff --git a/apps/backend/app/storage/local.py b/apps/backend/app/storage/local.py index 0b12199..5fde18a 100644 --- a/apps/backend/app/storage/local.py +++ b/apps/backend/app/storage/local.py @@ -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"] diff --git a/apps/backend/tests/test_ingestion_resource_budgets.py b/apps/backend/tests/test_ingestion_resource_budgets.py index 5821d75..00bba43 100644 --- a/apps/backend/tests/test_ingestion_resource_budgets.py +++ b/apps/backend/tests/test_ingestion_resource_budgets.py @@ -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 --------------------------------------