diff --git a/core/aria2_installer.py b/core/aria2_installer.py index 899ec8c..1c10d78 100644 --- a/core/aria2_installer.py +++ b/core/aria2_installer.py @@ -201,6 +201,14 @@ def _assert_within_install_root(path: Path) -> None: raise Aria2InstallError(f"Refusing to write outside aria2 install root: {target}") +def _validate_archive_member_path(destination_abs: Path, member_name: str) -> None: + target_abs = (destination_abs / member_name).resolve() + if target_abs != destination_abs and not is_path_within( + str(target_abs), str(destination_abs) + ): + raise Aria2InstallError(f"Unsafe path in aria2 archive: {member_name}") + + def _safe_extract_zip(archive_path: Path, destination: Path) -> None: destination_abs = destination.resolve() with zipfile.ZipFile(archive_path) as archive: @@ -210,10 +218,7 @@ def _safe_extract_zip(archive_path: Path, destination: Path) -> None: raise Aria2InstallError( f"Refusing symbolic link in aria2 archive: {member.filename}" ) - target = destination_abs / member.filename - target_abs = target.resolve() - if not is_path_within(str(target_abs), str(destination_abs)) and target_abs != destination_abs: - raise Aria2InstallError(f"Unsafe path in aria2 archive: {member.filename}") + _validate_archive_member_path(destination_abs, member.filename) archive.extractall(destination_abs) @@ -229,10 +234,7 @@ def _safe_extract_tar(archive_path: Path, destination: Path) -> None: raise Aria2InstallError( f"Refusing special file in aria2 archive: {member.name}" ) - target = destination_abs / member.name - target_abs = target.resolve() - if not is_path_within(str(target_abs), str(destination_abs)) and target_abs != destination_abs: - raise Aria2InstallError(f"Unsafe path in aria2 archive: {member.name}") + _validate_archive_member_path(destination_abs, member.name) archive.extractall(destination_abs) diff --git a/tests/test_path_security.py b/tests/test_path_security.py index 22246ec..3910423 100644 --- a/tests/test_path_security.py +++ b/tests/test_path_security.py @@ -1,15 +1,22 @@ import io import os +import stat import tarfile import tempfile import unittest +import zipfile from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch import requests -from core.aria2_installer import ARIA2_INSTALL_ROOT, Aria2InstallError, _safe_extract_tar +from core.aria2_installer import ( + ARIA2_INSTALL_ROOT, + Aria2InstallError, + _safe_extract_tar, + _safe_extract_zip, +) from core.download.api import context as downloader_module from core.download.api import ( download_file, @@ -722,6 +729,44 @@ class PathSecurityTests(unittest.TestCase): with self.assertRaises(Aria2InstallError): _safe_extract_tar(Path(archive_path), destination) + def test_zip_paths_outside_destination_are_rejected(self): + with tempfile.TemporaryDirectory() as temp_dir: + archive_path = Path(temp_dir) / "aria2.zip" + destination = Path(temp_dir) / "extract" + destination.mkdir() + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("../outside.txt", "unsafe") + + with self.assertRaises(Aria2InstallError): + _safe_extract_zip(archive_path, destination) + + def test_zip_links_are_rejected(self): + with tempfile.TemporaryDirectory() as temp_dir: + archive_path = Path(temp_dir) / "aria2.zip" + destination = Path(temp_dir) / "extract" + destination.mkdir() + link = zipfile.ZipInfo("escape") + link.create_system = 3 + link.external_attr = (stat.S_IFLNK | 0o777) << 16 + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr(link, "../../outside") + + with self.assertRaises(Aria2InstallError): + _safe_extract_zip(archive_path, destination) + + def test_tar_paths_outside_destination_are_rejected(self): + with tempfile.TemporaryDirectory() as temp_dir: + archive_path = Path(temp_dir) / "aria2.tar" + destination = Path(temp_dir) / "extract" + destination.mkdir() + with tarfile.open(archive_path, "w") as archive: + payload = tarfile.TarInfo("../outside.txt") + payload.size = 6 + archive.addfile(payload, io.BytesIO(b"unsafe")) + + with self.assertRaises(Aria2InstallError): + _safe_extract_tar(archive_path, destination) + if __name__ == "__main__": unittest.main()