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

@@ -16,6 +16,7 @@
     Chooser,
     Env,
     Factory,
+    Link,
     PoetryDependency,
     PoetryPackage,
     PoetryProjectPackage,
@@ -278,12 +279,27 @@ def parse_pip_requirement(requirement: str) -> Optional[Dict[str, str]]:
     return match.groupdict()
 
 
+class PoetryDependencyWithHash(PoetryDependency):
+    def __init__(self, *args, hash: Optional[str] = None, **kwargs) -> None:
+        self.hash = hash
+        super().__init__(*args, **kwargs)
+
+    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 +375,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 +398,37 @@ 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: 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 self._dependency_provides_hash():
+            return self.dependency.get_hash_model()
+        return None
+
+    def _dependency_provides_hash(self) -> bool:
+        return isinstance(self.dependency, PoetryDependencyWithHash)
+
+    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,
         )