Add installer hashes to info.json

constructor/_schema.py

@@ -858,7 +858,7 @@ class ConstructorConfiguration(BaseModel):
             It expects either a list of strings or single-key dictionaries.
 
             Allowed strings / keys: {}.
-            """.format(", ".join([f"`{v}`" for v in BuildOutputs.__members__.values()])),
+            """.format(", ".join(f"`{member.value}`" for member in BuildOutputs))
         ),
     )
     uninstall_with_conda_exe: bool | None = None

constructor/build_outputs.py

@@ -4,7 +4,6 @@
 Update documentation in `construct.py` if any changes are made.
 """
 
-import hashlib
 import json
 import logging
 import os
@@ -17,7 +16,9 @@
 from conda.exports import default_prefix
 
 from . import __version__
+from ._schema import BuildOutputs
 from .conda_interface import VersionOrder
+from .utils import hash_files
 
 logger = logging.getLogger(__name__)
 
@@ -59,48 +60,80 @@ def _validate_output(output):
     return {key: (value or {}) for (key, value) in output.items()}
 
 
-def process_build_outputs(info):
+def _needed_hash_algorithms(info: dict) -> set[str]:
+    """Return hash algorithms required by the requested build outputs."""
+    algorithms = set()
+
     for output in info.get("build_outputs", ()):
         output = _validate_output(output)
         name, config = output.popitem()
+
+        if name == BuildOutputs.INFO_JSON:
+            algorithms.add("sha256")
+        elif name == BuildOutputs.HASH:
+            algorithm = config.get("algorithm")
+            if isinstance(algorithm, str):
+                algorithms.add(algorithm)
+            elif algorithm:
+                algorithms.update(algorithm)
+
+    return algorithms
+
+
+def process_build_outputs(info: dict):
+    algorithms = _needed_hash_algorithms(info)
+
+    if algorithms:
+        info["_installer_hashes"] = hash_files(
+            [info["_outpath"]],
+            algorithms,
+        )
+
+    for output in info.get("build_outputs", ()):
+        output = _validate_output(output)
+
+        name, config = output.popitem()
+
         handler = OUTPUT_HANDLERS.get(name)
         if not handler:
             raise ValueError(
-                f"'output_builds' key {name} is not recognized! "
+                f"'build_outputs' key {name} is not recognized! "
                 f"Available keys: {tuple(OUTPUT_HANDLERS.keys())}"
             )
+
         outpath = handler(info, **config)
         if outpath:
             logger.info("build_outputs: '%s' created '%s'.", name, outpath)
 
 
-def dump_hash(info, algorithm=None):
+def dump_hash(info: dict, algorithm: str | None = None):
     if not algorithm:
         logger.warning("`hash` requires an algorithm. No hash files will be output.")
         return ""
+
     if isinstance(algorithm, str):
-        algorithm = [algorithm]
-    algorithms = set(algorithm)
-    if any(algo not in hashlib.algorithms_available for algo in algorithms):
-        invalid = algorithms.difference(set(hashlib.algorithms_available))
-        raise ValueError(f"Invalid algorithm: {', '.join(invalid)}")
-    BUFFER_SIZE = 65536
-    if isinstance(info["_outpath"], str):
-        installers = [Path(info["_outpath"])]
+        algorithms = [algorithm]
     else:
-        installers = [Path(outpath) for outpath in info["_outpath"]]
+        algorithms = algorithm
+
+    installer = Path(info["_outpath"])
     outpaths = []
-    for installer in installers:
-        filehashes = {algo: hashlib.new(algo) for algo in algorithms}
-        with open(installer, "rb") as f:
-            while buffer := f.read(BUFFER_SIZE):
-                for algo in algorithms:
-                    filehashes[algo].update(buffer)
-        for algo, filehash in filehashes.items():
-            outpath = Path(f"{installer}.{algo}")
-            with open(outpath, "w", newline="\n") as f:
-                f.write(f"{filehash.hexdigest()}  {installer.name}\n")
-            outpaths.append(str(outpath.absolute()))
+
+    for algo in algorithms:
+        try:
+            filehash = info["_installer_hashes"][algo]
+        except KeyError:
+            raise RuntimeError(
+                f"Hash for algorithm '{algo}' not found. "
+                f"Available algorithms: {', '.join(info.get('_installer_hashes', {}).keys())}"
+            ) from None
+        outpath = Path(f"{installer}.{algo}")
+
+        with open(outpath, "w", newline="\n") as f:
+            f.write(f"{filehash}  {installer.name}\n")
+
+        outpaths.append(str(outpath.absolute()))
+
     return ", ".join(outpaths)
 
 

constructor/conda_interface.py

@@ -155,8 +155,9 @@ def write_repodata(cache_dir, url, full_repodata, used_packages, info):
                 data = deepcopy(full_repodata[original_key][original_package])
                 pkg_fn = join(info["_download_dir"], package)
                 data["size"] = os.stat(pkg_fn).st_size
-                data["sha256"] = hash_files([pkg_fn], algorithm="sha256")
-                data["md5"] = hash_files([pkg_fn])
+                hashes = hash_files([pkg_fn], ["sha256", "md5"])
+                data["sha256"] = hashes["sha256"]
+                data["md5"] = hashes["md5"]
                 used_repodata[key][package] = data
 
         # In conda <23.1, the first line of the JSON should contain cache metadata

constructor/shar.py

@@ -89,7 +89,10 @@ def get_header(conda_exec, tarball, info):
     variables["installer_name"] = name
     variables["installer_version"] = info["version"]
     variables["installer_platform"] = info["_platform"]
-    variables["installer_md5"] = hash_files([conda_exec, *info["_internal_conda_files"], tarball])
+    variables["installer_md5"] = hash_files(
+        [conda_exec, *info["_internal_conda_files"], tarball],
+        "md5",
+    )["md5"]
     variables["default_prefix"] = info.get("default_prefix", "${HOME:-/opt}/%s" % name.lower())
     variables["first_payload_size"] = getsize(conda_exec)
     variables["second_payload_size"] = getsize(tarball)

constructor/utils.py

@@ -76,16 +76,46 @@ def replace(match):
     return pat.sub(replace, data)
 
 
-def hash_files(paths, algorithm="md5"):
-    h = hashlib.new(algorithm)
+def hash_files(paths: list[Path], algorithms: list[str] | str) -> dict[str, str]:
+    """
+    Calculate one or more hashes for the given files in a single pass.
+
+    Parameters
+    ----------
+    paths
+        An iterable of paths to hash.
+    algorithms
+        An iterable of hashlib algorithm names, such as ``md5`` or ``sha256``
+
+    Returns
+    -------
+    dict[str, str]
+        A mapping of algorithm names to digest values.
+    """
+    if isinstance(algorithms, str):
+        algorithms = [algorithms]
+
+    algorithms = set(algorithms)
+    invalid = algorithms.difference(hashlib.algorithms_available)
+    if invalid:
+        invalid_algo = "algorithm" if len(invalid) == 1 else "algorithms"
+        raise ValueError(f"Invalid {invalid_algo}: {', '.join(sorted(invalid))}")
+
+    BUFFER_SIZE = 65536
+
+    hashes = {algo: hashlib.new(algo) for algo in algorithms}
+
     for path in paths:
-        with open(path, "rb") as fi:
+        with open(path, "rb") as f:
             while True:
-                chunk = fi.read(262144)
+                chunk = f.read(BUFFER_SIZE)
                 if not chunk:
                     break
-                h.update(chunk)
-    return h.hexdigest()
+
+                for filehash in hashes.values():
+                    filehash.update(chunk)
+
+    return {algorithm: filehash.hexdigest() for algorithm, filehash in hashes.items()}
 
 
 def make_VIProductVersion(version):

news/1327-installer-hashes-to-info-json

@@ -0,0 +1,19 @@
+### Enhancements
+
+* Add installer hashes to `info.json` making it the single source of truth for installer checksums. (#1327)
+
+### Bug fixes
+
+* <news item>
+
+### Deprecations
+
+* <news item>
+
+### Docs
+
+* <news item>
+
+### Other
+
+* <news item>

tests/test_examples.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import getpass
+import hashlib
 import json
 import os
 import shutil
@@ -2027,6 +2028,16 @@ def test_output_files(tmp_path, installer_type):
         assert len(_records) > 0, f"Record for {env} is empty."
         assert isinstance(_records[0], dict), f"Record for {env} is not serialized."
 
+    expected_hashes = {
+        "sha256": hashlib.sha256((root_path / installer.name).read_bytes()).hexdigest(),
+        "md5": hashlib.md5((root_path / installer.name).read_bytes()).hexdigest(),
+    }
+    assert info_json.get("_installer_hashes") == expected_hashes
+    for algorithm, expected_hash in expected_hashes.items():
+        hash_file = root_path / f"{installer.name}.{algorithm}"
+        assert hash_file.exists()
+        assert hash_file.read_text() == f"{expected_hash}  {installer.name}\n"
+
 
 @pytest.mark.parametrize(
     "installer_type", installer_types_for_example(_example_path("regressions"))

tests/test_outputs.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from constructor.build_outputs import dump_hash
+from constructor.build_outputs import _needed_hash_algorithms, dump_hash
 
 TEST_FILES = {
     "test.txt": {
@@ -22,28 +22,48 @@
 @pytest.mark.parametrize(
     "algorithm,context",
     (
-        pytest.param("bad algorithm", pytest.raises(ValueError), id="invalid algorithm"),
         pytest.param("sha256", nullcontext(), id="string"),
         pytest.param(["sha256", "md5"], nullcontext(), id="list"),
     ),
 )
 def test_hash_dump(tmp_path, algorithm, context):
-    info = {"_outpath": []}
     for file, data in TEST_FILES.items():
         testfile = tmp_path / file
         testfile.write_text(data["content"])
-        info["_outpath"].append(str(testfile))
+    info = {
+        "_outpath": str(testfile),
+        "_installer_hashes": {algo: data[algo] for algo in ("sha256", "md5")},
+    }
     with context:
         dump_hash(info, algorithm=algorithm)
         if isinstance(algorithm, str):
             algorithm = [algorithm]
-        for file in info["_outpath"]:
-            for algo in algorithm:
-                hashfile = Path(f"{file}.{algo}")
-                assert hashfile.exists()
-                with open(hashfile, newline="") as f:
-                    content = f.read()
-                assert "\r" not in content
-                filehash, filename = content.strip().split()
-                assert filename == Path(file).name
-                assert filehash == TEST_FILES[filename][algo]
+        for algo in algorithm:
+            hashfile = Path(f"{testfile}.{algo}")
+            assert hashfile.exists()
+            with open(hashfile, newline="") as f:
+                content = f.read()
+            assert "\r" not in content
+            filehash, filename = content.strip().split()
+            assert filename == Path(file).name
+            assert filehash == TEST_FILES[filename][algo]
+
+
+@pytest.mark.parametrize(
+    "build_outputs, expected_algorithms",
+    (
+        pytest.param([], set(), id="neither info.json nor hash request"),
+        pytest.param(["info.json"], {"sha256"}, id="info.json only"),
+        pytest.param(
+            [{"hash": {"algorithm": "md5"}}], {"md5"}, id="no info.json, only md5 requested"
+        ),
+        pytest.param(
+            ["info.json", {"hash": {"algorithm": "md5"}}],
+            {"sha256", "md5"},
+            id="both info.json and md5 requested",
+        ),
+    ),
+)
+def test_hash_algorithms(build_outputs, expected_algorithms):
+    info = {"build_outputs": build_outputs}
+    assert _needed_hash_algorithms(info) == expected_algorithms

tests/test_utils.py

@@ -1,9 +1,12 @@
 from os import sep
 
+import pytest
+
 from constructor.utils import (
     bat_echo_esc,
     bat_env_var_esc,
     get_condarc_content,
+    hash_files,
     make_VIProductVersion,
     normalize_path,
 )
@@ -107,3 +110,12 @@ def test_get_condarc_content_returns_none():
     # write_condarc without channels should also return None
     info = {"write_condarc": True}
     assert get_condarc_content(info) is None
+
+
+def test_invalid_algorithm(tmp_path):
+    """Test that hash_files raises a ValueError for invalid algorithm names."""
+
+    path = tmp_path / "test.txt"
+    path.write_text("test string")
+    with pytest.raises(ValueError, match="bad_algorithm"):
+        hash_files([path], "bad_algorithm")