Skip to content
Closed
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
50 changes: 37 additions & 13 deletions cloakbrowser/download.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,12 +179,16 @@ def _verify_download_checksum(file_path: Path, version: str | None = None) -> No
tarball_name = get_archive_name()

if checksums is None:
logger.warning("SHA256SUMS not available for this release — skipping checksum verification")
logger.warning(
"SHA256SUMS not available for this release — skipping checksum verification"
)
return

expected = checksums.get(tarball_name)
if expected is None:
logger.warning("SHA256SUMS found but no entry for %s — skipping verification", tarball_name)
logger.warning(
"SHA256SUMS found but no entry for %s — skipping verification", tarball_name
)
return

_verify_checksum(file_path, expected)
Expand Down Expand Up @@ -247,7 +251,9 @@ def _download_file(url: str, dest: Path) -> None:
"""Download a file with progress logging."""
logger.info("Downloading from %s", url)

with httpx.stream("GET", url, follow_redirects=True, timeout=DOWNLOAD_TIMEOUT) as response:
with httpx.stream(
"GET", url, follow_redirects=True, timeout=DOWNLOAD_TIMEOUT
) as response:
response.raise_for_status()

total = int(response.headers.get("content-length", 0))
Expand Down Expand Up @@ -283,6 +289,7 @@ def _extract_archive(
# Clean existing dir if partial download existed
if dest_dir.exists():
import shutil

shutil.rmtree(dest_dir)

dest_dir.mkdir(parents=True, exist_ok=True)
Expand Down Expand Up @@ -310,21 +317,38 @@ def _extract_archive(
logger.info("Binary ready: %s", bp)


def _is_relative_to(path: Path, parent: Path) -> bool:
"""Return whether path is contained by parent after resolution."""
try:
path.relative_to(parent)
except ValueError:
return False
return True


def _extract_tar(archive_path: Path, dest_dir: Path) -> None:
"""Extract tar.gz archive with path traversal protection."""
resolved_dest = dest_dir.resolve()
with tarfile.open(archive_path, "r:gz") as tar:
safe_members = []
for member in tar.getmembers():
member_path = (dest_dir / member.name).resolve()
if not _is_relative_to(member_path, resolved_dest):
raise RuntimeError(f"Archive contains path traversal: {member.name}")

# Allow symlinks — macOS .app bundles require them (Framework layout)
if member.issym() or member.islnk():
link_target = member.linkname
if os.path.isabs(link_target) or ".." in link_target.split("/"):
logger.warning("Skipping suspicious symlink: %s -> %s", member.name, link_target)
link_path = (member_path.parent / link_target).resolve()
if os.path.isabs(link_target) or not _is_relative_to(
link_path, resolved_dest
):
logger.warning(
"Skipping suspicious symlink: %s -> %s",
member.name,
link_target,
)
continue
else:
member_path = (dest_dir / member.name).resolve()
if not str(member_path).startswith(str(dest_dir.resolve())):
raise RuntimeError(f"Archive contains path traversal: {member.name}")
safe_members.append(member)

tar.extractall(dest_dir, members=safe_members)
Expand All @@ -334,10 +358,11 @@ def _extract_zip(archive_path: Path, dest_dir: Path) -> None:
"""Extract zip archive with path traversal protection."""
import zipfile

resolved_dest = dest_dir.resolve()
with zipfile.ZipFile(archive_path, "r") as zf:
for info in zf.infolist():
member_path = (dest_dir / info.filename).resolve()
if not str(member_path).startswith(str(dest_dir.resolve())):
if not _is_relative_to(member_path, resolved_dest):
raise RuntimeError(f"Archive contains path traversal: {info.filename}")
zf.extractall(dest_dir)

Expand Down Expand Up @@ -419,6 +444,7 @@ def binary_info() -> dict:
# Auto-update
# ---------------------------------------------------------------------------


def check_for_update() -> str | None:
"""Manually check for a newer Chromium version. Returns new version or None.

Expand Down Expand Up @@ -470,9 +496,7 @@ def _get_latest_chromium_version() -> str | None:
so Linux-only releases won't be offered to macOS users.
"""
try:
resp = httpx.get(
GITHUB_API_URL, params={"per_page": 10}, timeout=10.0
)
resp = httpx.get(GITHUB_API_URL, params={"per_page": 10}, timeout=10.0)
resp.raise_for_status()
platform_tarball = get_archive_name()
for release in resp.json():
Expand Down
59 changes: 52 additions & 7 deletions tests/test_extract.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
"""Unit tests for archive extraction — path traversal protection, flattening, permissions."""

import io
import os
import platform
import stat
import tarfile
import zipfile
from pathlib import Path

import pytest

Expand All @@ -23,7 +22,7 @@
# ---------------------------------------------------------------------------


def _create_tar_gz(tmp_path, members: dict[str, bytes]) -> "Path":
def _create_tar_gz(tmp_path, members: dict[str, bytes]) -> Path:
"""Create a tar.gz with given {name: content} members."""
archive = tmp_path / "test.tar.gz"
with tarfile.open(archive, "w:gz") as tar:
Expand All @@ -36,7 +35,9 @@ def _create_tar_gz(tmp_path, members: dict[str, bytes]) -> "Path":

class TestExtractTar:
def test_basic(self, tmp_path):
archive = _create_tar_gz(tmp_path, {"chrome": b"binary", "lib/libfoo.so": b"lib"})
archive = _create_tar_gz(
tmp_path, {"chrome": b"binary", "lib/libfoo.so": b"lib"}
)
dest = tmp_path / "out"
dest.mkdir()
_extract_tar(archive, dest)
Expand All @@ -55,6 +56,35 @@ def test_path_traversal_blocked(self, tmp_path):
with pytest.raises(RuntimeError, match="path traversal"):
_extract_tar(archive, dest)

def test_sibling_prefix_traversal_blocked(self, tmp_path):
"""Containment should use path ancestry, not string prefixes."""
archive = tmp_path / "evil.tar.gz"
with tarfile.open(archive, "w:gz") as tar:
info = tarfile.TarInfo(name="../out_evil/owned.txt")
content = b"evil"
info.size = len(content)
tar.addfile(info, io.BytesIO(content))

dest = tmp_path / "out"
dest.mkdir()
with pytest.raises(RuntimeError, match="path traversal"):
_extract_tar(archive, dest)
assert not (tmp_path / "out_evil" / "owned.txt").exists()

def test_symlink_entry_path_traversal_blocked(self, tmp_path):
archive = tmp_path / "symlink-path.tar.gz"
with tarfile.open(archive, "w:gz") as tar:
sym = tarfile.TarInfo(name="../out_evil/link")
sym.type = tarfile.SYMTYPE
sym.linkname = "chrome"
tar.addfile(sym)

dest = tmp_path / "out"
dest.mkdir()
with pytest.raises(RuntimeError, match="path traversal"):
_extract_tar(archive, dest)
assert not (tmp_path / "out_evil" / "link").exists()

def test_suspicious_symlink_skipped(self, tmp_path):
"""Symlinks with absolute targets are skipped (logged as warning)."""
archive = tmp_path / "symlink.tar.gz"
Expand Down Expand Up @@ -83,7 +113,7 @@ def test_suspicious_symlink_skipped(self, tmp_path):
# ---------------------------------------------------------------------------


def _create_zip(tmp_path, members: dict[str, bytes]) -> "Path":
def _create_zip(tmp_path, members: dict[str, bytes]) -> Path:
"""Create a zip with given {name: content} members."""
archive = tmp_path / "test.zip"
with zipfile.ZipFile(archive, "w") as zf:
Expand All @@ -94,7 +124,9 @@ def _create_zip(tmp_path, members: dict[str, bytes]) -> "Path":

class TestExtractZip:
def test_basic(self, tmp_path):
archive = _create_zip(tmp_path, {"chrome.exe": b"binary", "lib/foo.dll": b"lib"})
archive = _create_zip(
tmp_path, {"chrome.exe": b"binary", "lib/foo.dll": b"lib"}
)
dest = tmp_path / "out"
dest.mkdir()
_extract_zip(archive, dest)
Expand All @@ -111,6 +143,17 @@ def test_path_traversal_blocked(self, tmp_path):
with pytest.raises(RuntimeError, match="path traversal"):
_extract_zip(archive, dest)

def test_sibling_prefix_traversal_blocked(self, tmp_path):
archive = tmp_path / "evil.zip"
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr("../out_evil/owned.txt", "evil")

dest = tmp_path / "out"
dest.mkdir()
with pytest.raises(RuntimeError, match="path traversal"):
_extract_zip(archive, dest)
assert not (tmp_path / "out_evil" / "owned.txt").exists()


# ---------------------------------------------------------------------------
# Directory flattening
Expand Down Expand Up @@ -169,7 +212,9 @@ def test_noop_multiple_entries(self, tmp_path):


class TestPermissions:
@pytest.mark.skipif(platform.system() == "Windows", reason="chmod not applicable on Windows")
@pytest.mark.skipif(
platform.system() == "Windows", reason="chmod not applicable on Windows"
)
def test_make_executable(self, tmp_path):
binary = tmp_path / "chrome"
binary.write_bytes(b"binary")
Expand Down