sec: ensure safe packages upon installation

Author: geritwagner

colrev/package_manager/package_manager.py

@@ -9,6 +9,10 @@
 import subprocess
 import sys
 import typing
+from pathlib import Path
+
+from packaging.requirements import InvalidRequirement
+from packaging.requirements import Requirement
 
 import colrev.exceptions as colrev_exceptions
 import colrev.package_manager.colrev_internal_packages
@@ -21,11 +25,41 @@
 
 def _validate_internal_package_selection(
     *, selected_package: str, internal_packages_dict: dict[str, str]
-) -> None:
+) -> str:
     if selected_package not in internal_packages_dict:
         raise ValueError(
             f"Installation rejected: unknown internal package '{selected_package}'."
         )
+    selected_path = (
+        Path(internal_packages_dict[selected_package]).expanduser().resolve()
+    )
+    if not selected_path.exists():
+        raise ValueError(
+            f"Installation rejected: internal package path does not exist: {selected_path}"
+        )
+    if not selected_path.is_dir():
+        raise ValueError(
+            f"Installation rejected: internal package path is not a directory: {selected_path}"
+        )
+    if not (selected_path / "pyproject.toml").is_file():
+        raise ValueError(
+            f"Installation rejected: internal package path misses pyproject.toml: {selected_path}"
+        )
+    return str(selected_path)
+
+
+def _validate_external_package_selection(*, selected_package: str) -> str:
+    if not selected_package:
+        raise ValueError("Installation rejected: empty package spec")
+    if selected_package.startswith("-"):
+        raise ValueError(
+            "Installation rejected: package specs must not start with '-' "
+            "(option injection risk)"
+        )
+
+    # Accept only normal Python requirement specifications.
+    Requirement(selected_package)
+    return selected_package
 
 
 def _run_uv_pip_list() -> subprocess.CompletedProcess[str]:
@@ -194,9 +228,12 @@ def install(
     ) -> None:
         """Install packages using uv if available, otherwise fallback to pip."""
         if uv:
-            package_manager = ["uv", "pip"]
+            uv_executable = shutil.which("uv")
+            if uv_executable is None:
+                raise FileNotFoundError("uv executable not found")
+            package_manager = [uv_executable, "pip"]
         else:
-            package_manager = ["pip"]
+            package_manager = [sys.executable, "-m", "pip"]
 
         # print(package_manager)
 
@@ -227,16 +264,27 @@ def install(
         if editable:
             install_args.append("--editable")
 
-        # Install both internal and external packages in a single command
-        all_packages = [
-            internal_packages_dict[p] if p in internal_packages_dict else p
+        safe_internal_packages = [
+            _validate_internal_package_selection(
+                selected_package=p, internal_packages_dict=internal_packages_dict
+            )
             for p in colrev_packages
-        ] + packages
+        ]
+        safe_external_packages = [
+            _validate_external_package_selection(selected_package=p) for p in packages
+        ]
 
-        for selected_package in all_packages:
+        for safe_selected_package in safe_internal_packages + safe_external_packages:
             try:
-                # install_args += all_packages
-                subprocess.run(install_args + [selected_package], check=True)
-            except Exception as e:
-                print(f"Installation failed: {selected_package}")
-                print(e)
+                subprocess.run(  # nosec B603 - package spec/path is validated above; shell=False.
+                    install_args + [safe_selected_package],
+                    check=True,
+                )
+            except (
+                subprocess.CalledProcessError,
+                FileNotFoundError,
+                ValueError,
+                InvalidRequirement,
+            ) as exc:
+                print(f"Installation failed: {safe_selected_package}")
+                print(exc)

tests/1_env/package_manager_security_test.py

@@ -3,12 +3,17 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+import sys
 from unittest.mock import MagicMock
 
 import pytest
+from packaging.requirements import InvalidRequirement
 
 from colrev.package_manager import check
 from colrev.package_manager.package_manager import PackageManager
+from colrev.package_manager.package_manager import _validate_external_package_selection
+from colrev.package_manager.package_manager import _validate_internal_package_selection
 
 
 @pytest.mark.parametrize(
@@ -38,20 +43,27 @@ def test_check_package_installed_validates_package_name_before_subprocess(
 
 
 def test_install_accepts_internal_package_and_calls_subprocess(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
+    internal_dir = tmp_path / "colrev-allowed"
+    internal_dir.mkdir()
+    (internal_dir / "pyproject.toml").write_text("[project]\nname='colrev-allowed'\n")
+
     run_mock = MagicMock()
     monkeypatch.setattr(
         "colrev.package_manager.package_manager.subprocess.run", run_mock
     )
     monkeypatch.setattr(
         "colrev.package_manager.package_manager.get_internal_packages_dict",
-        MagicMock(return_value={"colrev.allowed": "colrev-allowed"}),
+        MagicMock(return_value={"colrev.allowed": str(internal_dir)}),
     )
 
     PackageManager().install(packages=["colrev.allowed"], upgrade=False)
 
-    run_mock.assert_called_once_with(["pip", "install", "colrev-allowed"], check=True)
+    run_mock.assert_called_once_with(
+        [sys.executable, "-m", "pip", "install", str(internal_dir.resolve())],
+        check=True,
+    )
 
 
 def test_install_rejects_unknown_internal_package_before_subprocess(
@@ -70,3 +82,98 @@ def test_install_rejects_unknown_internal_package_before_subprocess(
         PackageManager().install(packages=["colrev.unknown"], upgrade=False)
 
     run_mock.assert_not_called()
+
+
+@pytest.mark.parametrize("package_spec", ["requests", "requests>=2.0"])
+def test_validate_external_package_accepts(package_spec: str) -> None:
+    assert (
+        _validate_external_package_selection(selected_package=package_spec)
+        == package_spec
+    )
+
+
+@pytest.mark.parametrize(
+    "package_spec", ["", "--index-url=https://example.com", "-r requirements.txt"]
+)
+def test_validate_external_package_rejects(package_spec: str) -> None:
+    with pytest.raises((ValueError, InvalidRequirement)):
+        _validate_external_package_selection(selected_package=package_spec)
+
+
+def test_validate_internal_package_selection_accepts(tmp_path: Path) -> None:
+    package_dir = tmp_path / "internal"
+    package_dir.mkdir()
+    (package_dir / "pyproject.toml").write_text("[project]\nname='internal'\n")
+
+    selected = _validate_internal_package_selection(
+        selected_package="colrev.allowed",
+        internal_packages_dict={"colrev.allowed": str(package_dir)},
+    )
+    assert selected == str(package_dir.resolve())
+
+
+def test_validate_internal_package_selection_rejects_missing_path(
+    tmp_path: Path,
+) -> None:
+    missing_dir = tmp_path / "missing"
+    with pytest.raises(ValueError):
+        _validate_internal_package_selection(
+            selected_package="colrev.allowed",
+            internal_packages_dict={"colrev.allowed": str(missing_dir)},
+        )
+
+
+def test_validate_internal_package_selection_rejects_missing_pyproject(
+    tmp_path: Path,
+) -> None:
+    package_dir = tmp_path / "internal"
+    package_dir.mkdir()
+    with pytest.raises(ValueError):
+        _validate_internal_package_selection(
+            selected_package="colrev.allowed",
+            internal_packages_dict={"colrev.allowed": str(package_dir)},
+        )
+
+
+def test_install_uses_uv_executable_when_enabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    run_mock = MagicMock()
+    monkeypatch.setattr(
+        "colrev.package_manager.package_manager.subprocess.run", run_mock
+    )
+    monkeypatch.setattr(
+        "colrev.package_manager.package_manager.shutil.which", lambda _: "/usr/bin/uv"
+    )
+    monkeypatch.setattr(
+        "colrev.package_manager.package_manager.get_internal_packages_dict",
+        MagicMock(return_value={}),
+    )
+
+    PackageManager().install(packages=["requests"], upgrade=False, uv=True)
+
+    run_mock.assert_called_once_with(
+        ["/usr/bin/uv", "pip", "install", "requests"],
+        check=True,
+    )
+
+
+def test_install_validates_external_packages_before_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    run_mock = MagicMock()
+    monkeypatch.setattr(
+        "colrev.package_manager.package_manager.subprocess.run", run_mock
+    )
+    monkeypatch.setattr(
+        "colrev.package_manager.package_manager.get_internal_packages_dict",
+        MagicMock(return_value={}),
+    )
+
+    with pytest.raises(ValueError, match="must not start with '-'"):
+        PackageManager().install(
+            packages=["--index-url=https://example.com"],
+            upgrade=False,
+        )
+
+    run_mock.assert_not_called()