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
70 changes: 52 additions & 18 deletions scripts/build_pack.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,28 +168,33 @@ def _safe_member_path(raw_name: str, strip_components: int) -> PurePosixPath | N
return stripped


def _portable_key(path: PurePosixPath) -> str:
return "/".join(unicodedata.normalize("NFC", part).casefold() for part in path.parts)
def _portable_key(path: PurePosixPath, *, case_sensitive: bool = False) -> str:
normalized = "/".join(unicodedata.normalize("NFC", part) for part in path.parts)
return normalized if case_sensitive else normalized.casefold()


def _register_path_shape(
relative: PurePosixPath,
kind: str,
kinds: dict[str, str],
parent_keys: set[str],
*,
case_sensitive: bool = False,
) -> None:
key = _portable_key(relative)
key = _portable_key(relative, case_sensitive=case_sensitive)
if key in kinds:
raise PackBuildError(f"duplicate archive member: {relative}")
ancestors = [PurePosixPath(*relative.parts[:index]) for index in range(1, len(relative.parts))]
for ancestor in ancestors:
ancestor_kind = kinds.get(_portable_key(ancestor))
ancestor_kind = kinds.get(_portable_key(ancestor, case_sensitive=case_sensitive))
if ancestor_kind is not None and ancestor_kind != "directory":
raise PackBuildError(f"archive path descends through a file: {relative}")
if kind != "directory" and key in parent_keys:
raise PackBuildError(f"archive file shadows an existing directory: {relative}")
kinds[key] = kind
parent_keys.update(_portable_key(ancestor) for ancestor in ancestors)
parent_keys.update(
_portable_key(ancestor, case_sensitive=case_sensitive) for ancestor in ancestors
)


def _resolve_tar_link_path(
Expand Down Expand Up @@ -250,7 +255,13 @@ def _copy_exact(source: BinaryIO, destination: Path, expected_size: int) -> None
raise PackBuildError("archive member exceeded its declared size")


def _extract_tar(archive: Path, payload: Path, strip_components: int) -> None:
def _extract_tar(
archive: Path,
payload: Path,
strip_components: int,
*,
case_sensitive_paths: bool = False,
) -> None:
entries: list[tuple[tarfile.TarInfo, PurePosixPath, str]] = []
kinds: dict[str, str] = {}
parent_keys: set[str] = set()
Expand All @@ -268,7 +279,13 @@ def _extract_tar(archive: Path, payload: Path, strip_components: int) -> None:
kind = "link"
else:
raise PackBuildError(f"unsupported archive member: {relative}")
_register_path_shape(relative, kind, kinds, parent_keys)
_register_path_shape(
relative,
kind,
kinds,
parent_keys,
case_sensitive=case_sensitive_paths,
)
if member.isfile():
if member.size < 0:
raise PackBuildError("archive member has a negative size")
Expand All @@ -291,13 +308,17 @@ def _extract_tar(archive: Path, payload: Path, strip_components: int) -> None:
destination.chmod(0o755)

by_key = {
_portable_key(relative): (member, relative, kind)
_portable_key(relative, case_sensitive=case_sensitive_paths): (
member,
relative,
kind,
)
for member, relative, kind in entries
}
resolving: set[str] = set()

def materialize(relative: PurePosixPath) -> Path:
key = _portable_key(relative)
key = _portable_key(relative, case_sensitive=case_sensitive_paths)
entry = by_key.get(key)
if entry is None:
raise PackBuildError(f"archive link target is missing: {relative}")
Expand Down Expand Up @@ -328,7 +349,11 @@ def materialize(relative: PurePosixPath) -> Path:
if kind == "link":
materialize(relative)

_validate_physical_tree(payload, compressed_bytes=archive.stat().st_size)
_validate_physical_tree(
payload,
compressed_bytes=archive.stat().st_size,
case_sensitive_paths=case_sensitive_paths,
)


def _zip_kind(info: zipfile.ZipInfo) -> str:
Expand Down Expand Up @@ -380,13 +405,18 @@ def _extract_zip(archive: Path, payload: Path, strip_components: int) -> None:
_validate_physical_tree(payload, compressed_bytes=archive.stat().st_size)


def _validate_physical_tree(root: Path, *, compressed_bytes: int = 0) -> None:
def _validate_physical_tree(
root: Path,
*,
compressed_bytes: int = 0,
case_sensitive_paths: bool = False,
) -> None:
seen: set[str] = set()
total_bytes = 0
count = 0
for path in sorted(root.rglob("*"), key=lambda item: item.as_posix().casefold()):
relative = path.relative_to(root)
key = relative.as_posix().casefold()
key = _portable_key(relative, case_sensitive=case_sensitive_paths)
if key in seen:
raise PackBuildError(f"case-insensitive duplicate extracted path: {relative}")
seen.add(key)
Expand Down Expand Up @@ -471,10 +501,17 @@ def extract_upstream(
archive_type: str,
strip_components: int,
payload: Path,
*,
case_sensitive_paths: bool = False,
) -> None:
payload.mkdir(parents=True, exist_ok=False)
if archive_type in {"tar.gz", "tar.xz"}:
_extract_tar(archive, payload, strip_components)
_extract_tar(
archive,
payload,
strip_components,
case_sensitive_paths=case_sensitive_paths,
)
elif archive_type == "zip":
_extract_zip(archive, payload, strip_components)
elif archive_type == "7z-sfx":
Expand Down Expand Up @@ -504,10 +541,6 @@ def extract_upstream(
executable,
"x",
"-y",
"-spf-",
"-snl-",
"-snh-",
"-sns-",
f"-o{payload}",
str(archive),
],
Expand All @@ -521,7 +554,7 @@ def extract_upstream(
raise PackBuildError(f"7-Zip extraction failed: {completed.stdout[-2000:]}")
else:
raise PackBuildError(f"unsupported upstream archive: {archive_type}")
_validate_physical_tree(payload)
_validate_physical_tree(payload, case_sensitive_paths=case_sensitive_paths)


def _probe_command(component_id: str, name: str, executable: Path) -> list[str]:
Expand Down Expand Up @@ -743,6 +776,7 @@ def build_pack(
component["archiveType"],
component["stripComponents"],
payload,
case_sensitive_paths=target.startswith("linux-"),
)
probe_payload(component_id, component["version"], component["executables"], payload)
collect_licenses(payload, pack_root / "licenses")
Expand Down
23 changes: 18 additions & 5 deletions scripts/generate_release.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,19 +143,23 @@ def _register_pack_path(
kind: str,
kinds: dict[str, str],
parent_keys: set[str],
*,
case_sensitive: bool,
) -> None:
key = _portable_key(relative)
key = _portable_key(relative, case_sensitive=case_sensitive)
if key in kinds:
raise ReleaseBuildError(f"pack has duplicate paths: {relative}")
ancestors = [PurePosixPath(*relative.parts[:index]) for index in range(1, len(relative.parts))]
for ancestor in ancestors:
ancestor_kind = kinds.get(_portable_key(ancestor))
ancestor_kind = kinds.get(_portable_key(ancestor, case_sensitive=case_sensitive))
if ancestor_kind is not None and ancestor_kind != "directory":
raise ReleaseBuildError(f"pack path descends through a file: {relative}")
if kind != "directory" and key in parent_keys:
raise ReleaseBuildError(f"pack file shadows an existing directory: {relative}")
kinds[key] = kind
parent_keys.update(_portable_key(ancestor) for ancestor in ancestors)
parent_keys.update(
_portable_key(ancestor, case_sensitive=case_sensitive) for ancestor in ancestors
)


def _read_member_bytes(archive: tarfile.TarFile, member: tarfile.TarInfo, limit: int) -> bytes:
Expand Down Expand Up @@ -211,6 +215,7 @@ def audit_pack_archive(
payload_hashes: dict[str, str] = {}
license_files = 0
member_names: list[str] = []
case_sensitive_paths = str(metadata["target"]).startswith("linux-")
with tarfile.open(path, mode="r:xz") as archive:
for index, member in enumerate(archive, 1):
if index > MAX_MEMBERS:
Expand All @@ -225,7 +230,13 @@ def audit_pack_archive(
kind = "file"
else:
raise ReleaseBuildError(f"pack contains an unsupported member: {path.name}")
_register_pack_path(relative, kind, kinds, parent_keys)
_register_pack_path(
relative,
kind,
kinds,
parent_keys,
case_sensitive=case_sensitive_paths,
)
member_names.append(relative.as_posix())
if source_date_epoch is not None:
if member.mtime != source_date_epoch or member.uid != 0 or member.gid != 0:
Expand Down Expand Up @@ -301,7 +312,9 @@ def audit_pack_archive(
raise ReleaseBuildError(f"pack manifest bin directory is invalid: {path.name}")
directory_name = "payload" if relative_directory == "." else "payload/" + relative_directory
directory = _safe_name(directory_name)
if kinds.get(_portable_key(directory)) != "directory" and not any(
if kinds.get(
_portable_key(directory, case_sensitive=case_sensitive_paths)
) != "directory" and not any(
name.startswith("./" + directory.as_posix() + "/") for name in payload_hashes
):
raise ReleaseBuildError(f"pack bin directory is missing: {path.name}")
Expand Down
41 changes: 40 additions & 1 deletion tests/test_pack_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import json
import subprocess
import tarfile
from pathlib import Path
from pathlib import Path, PurePosixPath

import pytest

Expand All @@ -14,6 +14,7 @@
_audit_7z_listing_text,
_deterministic_tar_xz,
_extract_tar,
_register_path_shape,
_verify_authenticode_signature,
collect_licenses,
sha256_file,
Expand Down Expand Up @@ -92,6 +93,44 @@ def test_tar_extraction_rejects_file_directory_shadowing(tmp_path: Path) -> None
_extract_tar(archive, tmp_path / "payload", 1)


def test_tar_extraction_preserves_case_distinct_linux_paths(tmp_path: Path) -> None:
archive = tmp_path / "input.tar.gz"
with tarfile.open(archive, "w:gz") as handle:
for name, data in (
("root/share/terminfo/E/Eterm", b"upper"),
("root/share/terminfo/e/eterm", b"lower"),
):
member = tarfile.TarInfo(name)
member.size = len(data)
handle.addfile(member, io.BytesIO(data))

with pytest.raises(PackBuildError, match="duplicate archive member"):
_extract_tar(archive, tmp_path / "portable", 1)

kinds: dict[str, str] = {}
parent_keys: set[str] = set()
for relative in (
PurePosixPath("share/terminfo/E/Eterm"),
PurePosixPath("share/terminfo/e/eterm"),
):
_register_path_shape(
relative,
"file",
kinds,
parent_keys,
case_sensitive=True,
)
assert len(kinds) == 2

linux_payload = tmp_path / "linux"
case_probe = tmp_path / "case-probe"
(case_probe / "E").mkdir(parents=True)
if not (case_probe / "e").exists():
_extract_tar(archive, linux_payload, 1, case_sensitive_paths=True)
assert (linux_payload / "share/terminfo/E/Eterm").read_bytes() == b"upper"
assert (linux_payload / "share/terminfo/e/eterm").read_bytes() == b"lower"


def test_git_sfx_listing_rejects_traversal_and_links() -> None:
with pytest.raises(PackBuildError, match="escapes payload"):
_audit_7z_listing_text("Path = ../../outside.exe\n")
Expand Down
Loading