diff --git a/scripts/release_nvattest_proof.py b/scripts/release_nvattest_proof.py index 6038c586b..99738c48d 100644 --- a/scripts/release_nvattest_proof.py +++ b/scripts/release_nvattest_proof.py @@ -370,33 +370,118 @@ def _default_run_package_install( def _default_observe_installed_distributions( env_python: Path, ) -> Sequence[Mapping[str, Any]]: + support_names = json.dumps(sorted(SUPPORT_DISTRIBUTION_NAMES)) script = r""" from __future__ import annotations import hashlib import importlib.metadata import json +import os +import re from pathlib import Path +SUPPORT_DISTRIBUTION_NAMES = set(__SUPPORT_DISTRIBUTION_NAMES__) + + +def normalize_distribution_name(value): + return re.sub(r"[-_.]+", "-", value).lower() + + +def metadata_name(dist): + try: + return dist.metadata.get("Name", "") or "" + except Exception: + return "" + + +def dist_info_name(raw_path): + if raw_path is None: + return "" + name = Path(str(raw_path)).name + if not name.endswith(".dist-info"): + return "" + stem = name.removesuffix(".dist-info") + parts = stem.rsplit("-", 1) + return parts[0] if len(parts) == 2 else stem + + +def distribution_name(dist, raw_path): + return metadata_name(dist) or dist_info_name(raw_path) + + +def is_relevant(name): + normalized = normalize_distribution_name(name) + return normalized in SUPPORT_DISTRIBUTION_NAMES or normalized.startswith("solstone") + + +def metadata_field(metadata, field): + prefix = f"{field}:" + for line in metadata.splitlines(): + if line.startswith(prefix): + return line.split(":", 1)[1].strip() + return "" + + +def failure(error, *, expected, actual, repair): + failures.append( + { + "actual": actual, + "error": error, + "expected": expected, + "repair": repair, + } + ) + + entries = [] +failures = [] +seen = set() for dist in importlib.metadata.distributions(): raw_path = getattr(dist, "_path", None) + raw_name = distribution_name(dist, raw_path) + name = normalize_distribution_name(raw_name) if raw_path is None: + if is_relevant(raw_name): + failure( + "nvattest installed distribution has no resolvable dist-info path", + expected=f"resolvable dist-info path for {name}", + actual=raw_name or "", + repair=( + "repair distribution " + f"{name} so it has a resolvable dist-info path" + ), + ) continue + key = os.path.realpath(str(raw_path)) + if key in seen: + continue + seen.add(key) metadata_path = Path(str(raw_path)) / "METADATA" try: metadata_bytes = metadata_path.read_bytes() - except OSError: + metadata = metadata_bytes.decode("utf-8") + except (OSError, UnicodeDecodeError) as exc: + if is_relevant(raw_name): + failure( + "nvattest installed distribution dist-info METADATA could not be read", + expected=f"readable dist-info METADATA for {name}", + actual=f"{metadata_path}: {type(exc).__name__}", + repair=( + "repair distribution " + f"{name}'s dist-info METADATA so it can be read" + ), + ) continue entries.append( { "metadata_sha256": hashlib.sha256(metadata_bytes).hexdigest(), - "name": dist.metadata.get("Name", ""), - "version": dist.version, + "name": metadata_field(metadata, "Name") or raw_name, + "version": metadata_field(metadata, "Version"), } ) -print(json.dumps(entries, sort_keys=True)) -""" +print(json.dumps({"entries": entries, "failures": failures}, sort_keys=True)) +""".replace("__SUPPORT_DISTRIBUTION_NAMES__", support_names) result = _run_command((str(env_python), "-c", script)) if result.exit_code != 0: raise NvattestProofError( @@ -415,24 +500,72 @@ print(json.dumps(entries, sort_keys=True)) [ _failure( "nvattest installed distribution metadata query emitted invalid JSON", - expected="JSON list of distribution metadata facts", + expected="JSON object with entries and failures", actual=str(exc), ) ] ) from exc - if not isinstance(payload, list) or not all( - isinstance(entry, Mapping) for entry in payload - ): + if not isinstance(payload, Mapping) or set(payload) != {"entries", "failures"}: raise NvattestProofError( [ _failure( "nvattest installed distribution metadata query payload is invalid", - expected="JSON list of objects", + expected="JSON object with entries and failures", actual=repr(payload), ) ] ) - return cast(Sequence[Mapping[str, Any]], payload) + query_failures = payload.get("failures") + if not isinstance(query_failures, list) or not all( + isinstance(entry, Mapping) for entry in query_failures + ): + raise NvattestProofError( + [ + _failure( + "nvattest installed distribution metadata query failure set is invalid", + expected="JSON list of failure objects", + actual=repr(query_failures), + ) + ] + ) + failures: list[Failure] = [] + for entry in query_failures: + if set(entry) != {"actual", "error", "expected", "repair"} or not all( + isinstance(entry.get(key), str) + for key in ("actual", "error", "expected", "repair") + ): + failures.append( + _failure( + "nvattest installed distribution metadata query failure is invalid", + expected="failure object with string actual, error, expected, repair", + actual=repr(entry), + ) + ) + continue + failures.append( + _failure( + str(entry["error"]), + expected=str(entry["expected"]), + actual=str(entry["actual"]), + repair=str(entry["repair"]), + ) + ) + if failures: + raise NvattestProofError(failures) + entries = payload.get("entries") + if not isinstance(entries, list) or not all( + isinstance(entry, Mapping) for entry in entries + ): + raise NvattestProofError( + [ + _failure( + "nvattest installed distribution metadata query entry set is invalid", + expected="JSON list of distribution metadata objects", + actual=repr(entries), + ) + ] + ) + return cast(Sequence[Mapping[str, Any]], entries) def _default_integrity_recheck( @@ -985,6 +1118,10 @@ def _installed_closure_payload( "nvattest installed closure distribution is duplicated", expected="one installed distribution per expected name", actual=name, + repair=( + "repair the proof environment so it contains one " + f"installed distribution named {name}, then {REPAIR}" + ), ) ) continue diff --git a/tests/test_release_nvattest_proof.py b/tests/test_release_nvattest_proof.py index 303919333..d895eda0b 100644 --- a/tests/test_release_nvattest_proof.py +++ b/tests/test_release_nvattest_proof.py @@ -6,8 +6,11 @@ from __future__ import annotations import copy import hashlib import json +import os import platform +import shlex import shutil +import sys import zipfile from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass @@ -351,6 +354,105 @@ def test_installed_closure_rejects_missing_support_distribution(tmp_path: Path) ) +def test_installed_distribution_observer_dedupes_lib64_dist_info_alias( + tmp_path: Path, +) -> None: + lib = tmp_path / "lib" + lib.mkdir() + dist_info = _write_installed_dist_info(lib, name="solstone", version=VERSION) + lib64 = tmp_path / "lib64" + lib64.symlink_to(lib, target_is_directory=True) + + observed = proof._default_observe_installed_distributions( + _python_with_pythonpath(tmp_path, (lib, lib64)) + ) + + assert [ + entry + for entry in observed + if proof._normalize_distribution_name(entry["name"]) == "solstone" + ] == [ + { + "metadata_sha256": hashlib.sha256( + (dist_info / "METADATA").read_bytes() + ).hexdigest(), + "name": "solstone", + "version": VERSION, + } + ] + + +def test_installed_closure_rejects_duplicate_distribution_realpaths( + tmp_path: Path, +) -> None: + one = tmp_path / "one" + two = tmp_path / "two" + one.mkdir() + two.mkdir() + first = _write_installed_dist_info(one, name="solstone", version=VERSION) + _write_installed_dist_info(two, name="solstone", version=VERSION) + observed = proof._default_observe_installed_distributions( + _python_with_pythonpath(tmp_path, (one, two)) + ) + + with pytest.raises(proof.NvattestProofError) as exc_info: + proof._installed_closure_payload( + observed, + expected_candidate_wheels=( + { + "metadata_sha256": hashlib.sha256( + (first / "METADATA").read_bytes() + ).hexdigest(), + "name": "solstone", + "version": VERSION, + "wheel": "CANDIDATE/solstone-1.0.0-py3-none-any.whl", + "wheel_bytes": 1, + "wheel_sha256": "0" * 64, + }, + ), + expected_support_distributions=(), + ) + + duplicate = next( + failure + for failure in exc_info.value.failures + if failure.error == "nvattest installed closure distribution is duplicated" + ) + assert duplicate.actual == "solstone" + assert duplicate.repair == ( + "repair the proof environment so it contains one installed distribution " + "named solstone, then regenerate the retained nvattest proof from the " + "original release inputs" + ) + + +def test_installed_distribution_observer_reports_relevant_unreadable_metadata( + tmp_path: Path, +) -> None: + root = tmp_path / "site" + root.mkdir() + _write_installed_dist_info(root, name="solstone", version=VERSION, metadata=False) + + with pytest.raises(proof.NvattestProofError) as exc_info: + proof._default_observe_installed_distributions( + _python_with_pythonpath(tmp_path, (root,)) + ) + + assert all( + failure.error != "nvattest installed closure is missing distribution" + for failure in exc_info.value.failures + ) + failure = exc_info.value.failures[0] + assert ( + failure.error + == "nvattest installed distribution dist-info METADATA could not be read" + ) + assert failure.expected == "readable dist-info METADATA for solstone" + assert failure.repair == ( + "repair distribution solstone's dist-info METADATA so it can be read" + ) + + def test_command_text_normalization_fails_closed_on_prefix_collision() -> None: env_root = Path("/tmp/abc") candidate_dir = Path("/tmp/candidate") @@ -1137,6 +1239,46 @@ def _write_support_wheels(path: Path) -> tuple[Path, ...]: ) +def _python_with_pythonpath(tmp_path: Path, roots: Sequence[Path]) -> Path: + wrapper = tmp_path / "python-with-pythonpath" + cwd = tmp_path / "python-cwd" + cwd.mkdir() + pythonpath = os.pathsep.join(str(root) for root in roots) + wrapper.write_text( + "\n".join( + ( + "#!/bin/sh", + f"cd {shlex.quote(str(cwd))}", + ( + f"PYTHONPATH={shlex.quote(pythonpath)} " + f'exec {shlex.quote(sys.executable)} -S "$@"' + ), + "", + ) + ), + encoding="utf-8", + ) + wrapper.chmod(0o755) + return wrapper + + +def _write_installed_dist_info( + root: Path, + *, + name: str, + version: str, + metadata: bool = True, +) -> Path: + dist_info = root / f"{name.replace('-', '_')}-{version}.dist-info" + dist_info.mkdir() + if metadata: + (dist_info / "METADATA").write_text( + f"Name: {name}\nVersion: {version}\n", + encoding="utf-8", + ) + return dist_info + + def _write_metadata_wheel( path: Path, *,