Add support for pip hash checking

conda_lock/interfaces/vendored_poetry.py

@@ -5,6 +5,7 @@
 )
 from conda_lock._vendor.poetry.core.packages import URLDependency as PoetryURLDependency
 from conda_lock._vendor.poetry.core.packages import VCSDependency as PoetryVCSDependency
+from conda_lock._vendor.poetry.core.packages.utils.link import Link
 from conda_lock._vendor.poetry.factory import Factory
 from conda_lock._vendor.poetry.installation.chooser import Chooser
 from conda_lock._vendor.poetry.installation.operations.uninstall import Uninstall
@@ -21,6 +22,7 @@
     "Chooser",
     "Env",
     "Factory",
+    "Link",
     "PoetryDependency",
     "PoetryPackage",
     "PoetryProjectPackage",

conda_lock/models/lock_spec.py

@@ -29,6 +29,7 @@ class VersionedDependency(_BaseDependency):
     version: str
     build: Optional[str] = None
     conda_channel: Optional[str] = None
+    hash: Optional[str] = None
 
 
 class URLDependency(_BaseDependency):

conda_lock/pypi_solver.py

@@ -4,18 +4,20 @@
 
 from pathlib import Path
 from posixpath import expandvars
-from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, FrozenSet, List, Literal, Optional, Tuple, Union
 from urllib.parse import urldefrag, urlsplit, urlunsplit
 
 from clikit.api.io.flags import VERY_VERBOSE
 from clikit.io import ConsoleIO, NullIO
 from packaging.tags import compatible_tags, cpython_tags, mac_platforms
 from packaging.version import Version
 
+from conda_lock._vendor.poetry.core.semver import VersionConstraint
 from conda_lock.interfaces.vendored_poetry import (
     Chooser,
     Env,
     Factory,
+    Link,
     PoetryDependency,
     PoetryPackage,
     PoetryProjectPackage,
@@ -278,12 +280,51 @@ def parse_pip_requirement(requirement: str) -> Optional[Dict[str, str]]:
     return match.groupdict()
 
 
+class PoetryDependencyWithHash(PoetryDependency):
+    def __init__(
+        self,
+        name,  # type: str
+        constraint,  # type: Union[str, VersionConstraint]
+        optional=False,  # type: bool
+        category="main",  # type: str
+        allows_prereleases=False,  # type: bool
+        extras=None,  # type: Optional[Union[List[str], FrozenSet[str]]]
+        source_type=None,  # type: Optional[str]
+        source_url=None,  # type: Optional[str]
+        source_reference=None,  # type: Optional[str]
+        source_resolved_reference=None,  # type: Optional[str]
+        hash: Optional[str] = None,
+    ) -> None:
+        super().__init__(
+            name,
+            constraint,
+            optional=optional,
+            category=category,
+            allows_prereleases=allows_prereleases,
+            extras=extras,  # type: ignore  # upstream type hint is wrong
+            source_type=source_type,
+            source_url=source_url,
+            source_reference=source_reference,
+            source_resolved_reference=source_resolved_reference,
+        )
+        self.hash = hash
+
+    def get_hash_model(self) -> Optional[HashModel]:
+        if self.hash:
+            algo, value = self.hash.split(":")
+            return HashModel(**{algo: value})
+        return None
+
+
 def get_dependency(dep: lock_spec.Dependency) -> PoetryDependency:
     # FIXME: how do deal with extras?
     extras: List[str] = []
     if isinstance(dep, lock_spec.VersionedDependency):
-        return PoetryDependency(
-            name=dep.name, constraint=dep.version or "*", extras=dep.extras
+        return PoetryDependencyWithHash(
+            name=dep.name,
+            constraint=dep.version or "*",
+            extras=dep.extras,
+            hash=dep.hash,
         )
     elif isinstance(dep, lock_spec.URLDependency):
         return PoetryURLDependency(
@@ -359,14 +400,9 @@ def get_requirements(
             # https://github.com/conda/conda-lock/blob/ac31f5ddf2951ed4819295238ccf062fb2beb33c/conda_lock/_vendor/poetry/installation/executor.py#L557
             else:
                 link = chooser.choose_for(op.package)
-                parsed_url = urlsplit(link.url)
-                link.url = link.url.replace(parsed_url.netloc, str(parsed_url.hostname))
-                url = link.url_without_fragment
-                hashes: Dict[str, str] = {}
-                if link.hash_name is not None and link.hash is not None:
-                    hashes[link.hash_name] = link.hash
-                hash = HashModel.parse_obj(hashes)
-
+                url = _get_url(link)
+                hash_chooser = _HashChooser(link, op.package.dependency)
+                hash = hash_chooser.get_hash()
             if source_repository:
                 url = source_repository.normalize_solver_url(url)
 
@@ -387,6 +423,34 @@ def get_requirements(
     return requirements
 
 
+def _get_url(link: Link) -> str:
+    parsed_url = urlsplit(link.url)
+    link.url = link.url.replace(parsed_url.netloc, str(parsed_url.hostname))
+    return link.url_without_fragment
+
+
+class _HashChooser:
+    def __init__(
+        self, link: Link, dependency: Union[PoetryDependency, PoetryDependencyWithHash]
+    ):
+        self.link = link
+        self.dependency = dependency
+
+    def get_hash(self) -> HashModel:
+        return self._get_hash_from_dependency() or self._get_hash_from_link()
+
+    def _get_hash_from_dependency(self) -> Optional[HashModel]:
+        if isinstance(self.dependency, PoetryDependencyWithHash):
+            return self.dependency.get_hash_model()
+        return None
+
+    def _get_hash_from_link(self) -> HashModel:
+        hashes: Dict[str, str] = {}
+        if self.link.hash_name is not None and self.link.hash is not None:
+            hashes[self.link.hash_name] = self.link.hash
+        return HashModel.parse_obj(hashes)
+
+
 def solve_pypi(
     pip_specs: Dict[str, lock_spec.Dependency],
     use_latest: List[str],

conda_lock/src_parser/pyproject_toml.py

@@ -377,9 +377,25 @@ def to_match_spec(conda_dep_name: str, conda_version: Optional[str]) -> str:
     return spec
 
 
+class RequirementWithHash(Requirement):
+    """Requirement with support for pip hash checking.
+
+    Pip offers hash checking where the requirement string is
+    my_package == 1.23 --hash=sha256:1234...
+    """
+
+    def __init__(self, requirement_string: str) -> None:
+        try:
+            requirement_string, hash = requirement_string.split(" --hash=")
+        except ValueError:
+            hash = None
+        self.hash: Optional[str] = hash
+        super().__init__(requirement_string)
+
+
 def parse_requirement_specifier(
     requirement: str,
-) -> Requirement:
+) -> RequirementWithHash:
     """Parse a url requirement to a conda spec"""
     if (
         requirement.startswith("git+")
@@ -392,9 +408,9 @@ def parse_requirement_specifier(
         if repo_name.endswith(".git"):
             repo_name = repo_name[:-4]
         # Use the repo name as a placeholder for the package name
-        return Requirement(f"{repo_name} @ {requirement}")
+        return RequirementWithHash(f"{repo_name} @ {requirement}")
     else:
-        return Requirement(requirement)
+        return RequirementWithHash(requirement)
 
 
 def unpack_git_url(url: str) -> Tuple[str, Optional[str]]:
@@ -460,6 +476,7 @@ def parse_python_requirement(
             manager=manager,
             category=category,
             extras=extras,
+            hash=parsed_req.hash,
         )
 
 

tests/test-pip-hash-checking/environment.yml

@@ -0,0 +1,8 @@
+# environment.yml
+channels:
+  - conda-forge
+
+dependencies:
+  - pip
+  - pip:
+    - flit-core === 3.9.0 --hash=sha256:1234

tests/test_conda_lock.py

@@ -156,6 +156,13 @@ def pip_environment_different_names_same_deps(tmp_path: Path):
     )
 
 
+@pytest.fixture
+def pip_hash_checking_environment(tmp_path: Path):
+    return clone_test_dir("test-pip-hash-checking", tmp_path).joinpath(
+        "environment.yml"
+    )
+
+
 @pytest.fixture
 def pip_local_package_environment(tmp_path: Path):
     return clone_test_dir("test-local-pip", tmp_path).joinpath("environment.yml")
@@ -1508,6 +1515,22 @@ def test_run_lock_with_pip_environment_different_names_same_deps(
     run_lock([pip_environment_different_names_same_deps], conda_exe=conda_exe)
 
 
+def test_run_lock_with_pip_hash_checking(
+    monkeypatch: "pytest.MonkeyPatch",
+    pip_hash_checking_environment: Path,
+    conda_exe: str,
+):
+    work_dir = pip_hash_checking_environment.parent
+    monkeypatch.chdir(work_dir)
+    if is_micromamba(conda_exe):
+        monkeypatch.setenv("CONDA_FLAGS", "-v")
+    run_lock([pip_hash_checking_environment], conda_exe=conda_exe)
+
+    lockfile = parse_conda_lock_file(work_dir / DEFAULT_LOCKFILE_NAME)
+    hashes = {package.name: package.hash for package in lockfile.package}
+    assert hashes["flit-core"].sha256 == "1234"
+
+
 def test_run_lock_uppercase_pip(
     monkeypatch: "pytest.MonkeyPatch",
     env_with_uppercase_pip: Path,