Attempt to split hashes into a struct

conda_lock/conda_lock.py

@@ -495,15 +495,15 @@ def format_pip_requirement(
         if spec.source and spec.source.type == "url":
             return f"{spec.name} @ {spec.source.url}"
         elif direct:
-            return f'{spec.name} @ {spec.url}#{spec.hash.replace(":", "=")}'
+            return f"{spec.name} @ {spec.url}#md5={spec.hash.md5}"
         else:
-            return f"{spec.name} === {spec.version} --hash={spec.hash}"
+            return f"{spec.name} === {spec.version} --hash=md5:{spec.hash.md5}"
 
     def format_conda_requirement(
         spec: LockedDependency, platform: str, direct=False
     ) -> str:
         if direct:
-            return f"{spec.url}#{spec.hash.replace('md5:', '')}"
+            return f"{spec.url}#{spec.hash.md5}"
         else:
             path = pathlib.Path(urlsplit(spec.url).path)
             while path.suffix in {".tar", ".bz2", ".gz", ".conda"}:

conda_lock/conda_solver.py

@@ -30,6 +30,7 @@
 )
 from conda_lock.src_parser import (
     Dependency,
+    HashModel,
     LockedDependency,
     VersionedDependency,
     _apply_categories,
@@ -47,6 +48,7 @@ class FetchAction(TypedDict):
     depends: Optional[List[str]]
     fn: str
     md5: str
+    sha256: Optional[str]
     name: str
     subdir: str
     timestamp: int
@@ -152,7 +154,10 @@ def solve_conda(
             },
             url=action["url"],
             # NB: virtual packages may have no hash
-            hash=f"md5:{action['md5']}" if "md5" in action else "",
+            hash=HashModel(
+                md5=action["md5"] if "md5" in action else "",
+                sha256=action.get("sha256"),
+            ),
         )
         for action in dry_run_install["actions"]["FETCH"]
     }
@@ -230,7 +235,9 @@ def _reconstruct_fetch_actions(
                 "timestamp": item["timestamp"],
                 "url": item["url"],
                 "version": item["version"],
+                "sha256": item.get("sha256"),
             }
+
             dry_run_install["actions"]["FETCH"].append(repodata)
     return dry_run_install
 
@@ -433,14 +440,18 @@ def update_specs_for_arch(
             else:
                 channel = f'{entry["base_url"]}/{entry["platform"]}'
             url = f"{channel}/{fn}"
-            md5 = locked[package].hash
+            md5 = locked[package].hash.md5
+            if md5 is None:
+                raise RuntimeError("Conda packages require non-null md5 hashes")
+            sha256 = locked[package].hash.sha256
             dryrun_install["actions"]["FETCH"].append(
                 {
                     "name": entry["name"],
                     "channel": channel,
                     "url": url,
                     "fn": fn,
                     "md5": md5,
+                    "sha256": sha256,
                     "version": entry["version"],
                     "depends": [
                         f"{k} {v}".strip()
@@ -491,13 +502,16 @@ def fake_conda_environment(locked: Iterable[LockedDependency], platform: str):
                 "name": dep.name,
                 "channel": channel,
                 "url": dep.url,
-                "md5": dep.hash,
+                "md5": dep.hash.md5,
                 "build": build,
                 "build_number": build_number,
                 "version": dep.version,
                 "subdir": path.parent.name,
                 "depends": [f"{k} {v}".strip() for k, v in dep.dependencies.items()],
             }
+            if dep.hash.sha256 is not None:
+                entry["sha256"] = dep.hash.sha256
+
             with open(conda_meta / (path.name + ".json"), "w") as f:
                 json.dump(entry, f, indent=2)
         yield prefix

conda_lock/pypi_solver.py

@@ -268,15 +268,16 @@ def solve_pypi(
             source: Optional[src_parser.DependencySource] = None
             if op.package.source_type == "url":
                 url, fragment = urldefrag(op.package.source_url)
-                hash = fragment.replace("=", ":")
+                hash_type, hash = fragment.split("=")
+                hash = src_parser.HashModel(**{hash_type: hash})
                 source = src_parser.DependencySource(
                     type="url", url=op.package.source_url
                 )
             # Choose the most specific distribution for the target
             else:
                 link = chooser.choose_for(op.package)
                 url = link.url_without_fragment
-                hash = f"{link.hash_name}:{link.hash}"
+                hash = src_parser.HashModel(**{link.hash_name: link.hash})
 
             requirements.append(
                 src_parser.LockedDependency(

conda_lock/src_parser/__init__.py

@@ -75,14 +75,19 @@ class DependencySource(StrictModel):
 LockKey = namedtuple("LockKey", ["manager", "name", "platform"])
 
 
+class HashModel(StrictModel):
+    md5: Optional[str] = None
+    sha256: Optional[str] = None
+
+
 class LockedDependency(StrictModel):
     name: str
     version: str
     manager: Literal["conda", "pip"]
     platform: str
     dependencies: Dict[str, str] = {}
     url: str
-    hash: str
+    hash: HashModel
     optional: bool = False
     category: str = "main"
     source: Optional[DependencySource] = None
@@ -92,12 +97,7 @@ def key(self) -> LockKey:
 
     @validator("hash")
     def validate_hash(cls, v, values, **kwargs):
-        if ":" not in v:
-            if values["manager"] == "conda":
-                return f"md5:{v}"
-            raise ValueError("hash must specify an algorithm")
-        algorithm = v.split(":")[0]
-        if values["manager"] == "conda" and algorithm != "md5":
+        if (values["manager"] == "conda") and (v.md5 is None):
             raise ValueError("conda package hashes must use MD5")
         return v
 

conda_lock/src_parser/lockfile.py

@@ -4,7 +4,7 @@
 
 import yaml
 
-from . import Lockfile
+from . import LockedDependency, Lockfile
 
 
 def parse_conda_lock_file(
@@ -19,7 +19,7 @@ def parse_conda_lock_file(
     if not (isinstance(version, int) and version <= Lockfile.version):
         raise ValueError(f"{path} has unknown version {version}")
 
-    return Lockfile(**content)
+    return Lockfile.parse_obj(content)
 
 
 def write_conda_lock_file(