refactor: share archive member path validation

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)
 
 

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()