|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Qualify authoritative PyPI project-root metadata and package resolution. |
| 3 | +
|
| 4 | +This audit deliberately uses the PyPI JSON API and pip. Rendered project-page |
| 5 | +HTML is a separate, non-authoritative presentation surface and is not fetched |
| 6 | +here because an external client challenge must not be mistaken for metadata. |
| 7 | +""" |
| 8 | + |
| 9 | +from __future__ import annotations |
| 10 | + |
| 11 | +import argparse |
| 12 | +import json |
| 13 | +import os |
| 14 | +import re |
| 15 | +import shlex |
| 16 | +import subprocess |
| 17 | +import sys |
| 18 | +import tempfile |
| 19 | +import time |
| 20 | +import urllib.error |
| 21 | +import urllib.request |
| 22 | +from pathlib import Path |
| 23 | +from typing import Any |
| 24 | + |
| 25 | +try: |
| 26 | + from scripts.check_release_metadata import ReleaseMetadataError, SourceMetadata, load_source_metadata |
| 27 | +except ModuleNotFoundError as error: # pragma: no cover - direct command-line execution |
| 28 | + if error.name != "scripts": |
| 29 | + raise |
| 30 | + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
| 31 | + from scripts.check_release_metadata import ReleaseMetadataError, SourceMetadata, load_source_metadata |
| 32 | + |
| 33 | +PUBLIC_PYPI_INDEX = "https://pypi.org/simple" |
| 34 | +LEGACY_STABLE_VERSION = re.compile(r"0(?:\.[0-9]+)+") |
| 35 | + |
| 36 | + |
| 37 | +class ProjectSurfaceError(RuntimeError): |
| 38 | + """The canonical PyPI project surface differs from release authority.""" |
| 39 | + |
| 40 | + |
| 41 | +def _normalized_name(value: object) -> str: |
| 42 | + return re.sub(r"[-_.]+", "-", str(value or "").strip().lower()) |
| 43 | + |
| 44 | + |
| 45 | +def _release_files(releases: dict[str, Any], version: str) -> list[dict[str, Any]]: |
| 46 | + files = releases.get(version) |
| 47 | + if not isinstance(files, list) or not files or not all(isinstance(item, dict) for item in files): |
| 48 | + raise ProjectSurfaceError(f"PyPI release history does not retain files for {version}") |
| 49 | + return files |
| 50 | + |
| 51 | + |
| 52 | +def _legacy_version_key(version: str) -> tuple[int, ...]: |
| 53 | + return tuple(int(part) for part in version.split(".")) |
| 54 | + |
| 55 | + |
| 56 | +def verify_project_json(payload: object, source: SourceMetadata) -> str: |
| 57 | + """Verify root JSON and return the newest retained legacy stable version.""" |
| 58 | + |
| 59 | + if not isinstance(payload, dict): |
| 60 | + raise ProjectSurfaceError("PyPI project-root JSON must be an object") |
| 61 | + info = payload.get("info") |
| 62 | + releases = payload.get("releases") |
| 63 | + urls = payload.get("urls") |
| 64 | + if not isinstance(info, dict) or not isinstance(releases, dict) or not isinstance(urls, list): |
| 65 | + raise ProjectSurfaceError("PyPI project-root JSON lacks info, releases, or selected-release files") |
| 66 | + |
| 67 | + expected: dict[str, Any] = { |
| 68 | + "name": source.name, |
| 69 | + "version": source.registry_version, |
| 70 | + "summary": source.summary, |
| 71 | + "classifiers": list(source.classifiers), |
| 72 | + "description": source.readme, |
| 73 | + "description_content_type": "text/markdown", |
| 74 | + } |
| 75 | + for field, value in expected.items(): |
| 76 | + if info.get(field) != value: |
| 77 | + raise ProjectSurfaceError( |
| 78 | + f"authoritative PyPI project-root field {field} differs from current prerelease metadata" |
| 79 | + ) |
| 80 | + |
| 81 | + current_files = _release_files(releases, source.registry_version) |
| 82 | + if not all(isinstance(item, dict) for item in urls) or not urls: |
| 83 | + raise ProjectSurfaceError("PyPI project root does not expose current prerelease files") |
| 84 | + current_urls = {item.get("url") for item in current_files if isinstance(item.get("url"), str)} |
| 85 | + selected_urls = {item.get("url") for item in urls if isinstance(item.get("url"), str)} |
| 86 | + if not current_urls or selected_urls != current_urls: |
| 87 | + raise ProjectSurfaceError("PyPI project root selected-release files differ from the current prerelease") |
| 88 | + if any(item.get("yanked") is not False for item in current_files) or any( |
| 89 | + item.get("yanked") is not False for item in urls |
| 90 | + ): |
| 91 | + raise ProjectSurfaceError( |
| 92 | + f"current prerelease {source.registry_version} must remain installable and non-yanked" |
| 93 | + ) |
| 94 | + |
| 95 | + legacy_versions = sorted( |
| 96 | + (version for version in releases if isinstance(version, str) and LEGACY_STABLE_VERSION.fullmatch(version)), |
| 97 | + key=_legacy_version_key, |
| 98 | + ) |
| 99 | + if not legacy_versions: |
| 100 | + raise ProjectSurfaceError("PyPI project-root JSON no longer retains historical stable 0.x releases") |
| 101 | + retirement_required: list[str] = [] |
| 102 | + retirement_reason_required: list[str] = [] |
| 103 | + for version in legacy_versions: |
| 104 | + files = _release_files(releases, version) |
| 105 | + if any(file.get("yanked") is not True for file in files): |
| 106 | + retirement_required.append(version) |
| 107 | + elif any(not isinstance(file.get("yanked_reason"), str) or not file["yanked_reason"].strip() for file in files): |
| 108 | + retirement_reason_required.append(version) |
| 109 | + |
| 110 | + failures: list[str] = [] |
| 111 | + if retirement_required: |
| 112 | + failures.append( |
| 113 | + "yank these historical stable releases in PyPI release management with a retirement reason: " |
| 114 | + + ", ".join(retirement_required) |
| 115 | + ) |
| 116 | + if retirement_reason_required: |
| 117 | + failures.append("add a PyPI yank reason for: " + ", ".join(retirement_reason_required)) |
| 118 | + if failures: |
| 119 | + raise ProjectSurfaceError("; ".join(failures)) |
| 120 | + |
| 121 | + return legacy_versions[-1] |
| 122 | + |
| 123 | + |
| 124 | +def supported_prerelease_requirement(source: SourceMetadata) -> str: |
| 125 | + section = re.search(r"(?ms)^## Install\s*$\n(?P<body>.*?)(?=^##\s|\Z)", source.readme) |
| 126 | + if section is None: |
| 127 | + raise ProjectSurfaceError("README has no Install section") |
| 128 | + block = re.search(r"(?ms)^```(?:bash|shell)\s*$\n(?P<code>.*?)^```\s*$", section.group("body")) |
| 129 | + if block is None: |
| 130 | + raise ProjectSurfaceError("README Install section has no shell command") |
| 131 | + commands = [line.strip() for line in block.group("code").splitlines() if line.strip()] |
| 132 | + if not commands: |
| 133 | + raise ProjectSurfaceError("README Install section has an empty shell block") |
| 134 | + arguments = shlex.split(commands[0]) |
| 135 | + if len(arguments) != 4 or arguments[:3] != ["pip", "install", "--pre"]: |
| 136 | + raise ProjectSurfaceError("README first install command must explicitly select a prerelease requirement") |
| 137 | + requirement = arguments[3] |
| 138 | + if requirement != f"{source.name}~=2.0.0rc0": |
| 139 | + raise ProjectSurfaceError("README first install command must select the supported 2.0 prerelease line") |
| 140 | + return requirement |
| 141 | + |
| 142 | + |
| 143 | +def verify_pip_report(report: object, source: SourceMetadata, expected_version: str) -> None: |
| 144 | + if not isinstance(report, dict) or not isinstance(report.get("install"), list): |
| 145 | + raise ProjectSurfaceError("pip resolution report is invalid") |
| 146 | + selected = next( |
| 147 | + ( |
| 148 | + item |
| 149 | + for item in report["install"] |
| 150 | + if isinstance(item, dict) |
| 151 | + and isinstance(item.get("metadata"), dict) |
| 152 | + and _normalized_name(item["metadata"].get("name")) == _normalized_name(source.name) |
| 153 | + ), |
| 154 | + None, |
| 155 | + ) |
| 156 | + if selected is None: |
| 157 | + raise ProjectSurfaceError(f"pip did not select {source.name}") |
| 158 | + version = selected["metadata"].get("version") |
| 159 | + if version != expected_version: |
| 160 | + raise ProjectSurfaceError(f"pip selected {version or '<missing>'}; expected {expected_version}") |
| 161 | + |
| 162 | + |
| 163 | +def _pip_report(requirement: str, *, prerelease: bool) -> object: |
| 164 | + with tempfile.TemporaryDirectory(prefix="dw-pypi-project-surface-") as temporary: |
| 165 | + report_path = Path(temporary) / "pip-report.json" |
| 166 | + command = [ |
| 167 | + sys.executable, |
| 168 | + "-m", |
| 169 | + "pip", |
| 170 | + "install", |
| 171 | + "--disable-pip-version-check", |
| 172 | + "--dry-run", |
| 173 | + "--ignore-installed", |
| 174 | + "--no-deps", |
| 175 | + "--index-url", |
| 176 | + PUBLIC_PYPI_INDEX, |
| 177 | + ] |
| 178 | + if prerelease: |
| 179 | + command.append("--pre") |
| 180 | + command.extend(("--report", str(report_path), requirement)) |
| 181 | + environment = { |
| 182 | + **os.environ, |
| 183 | + "PIP_CONFIG_FILE": os.devnull, |
| 184 | + "PIP_INDEX_URL": PUBLIC_PYPI_INDEX, |
| 185 | + "PYTHONPATH": "", |
| 186 | + } |
| 187 | + for variable in ("PIP_EXTRA_INDEX_URL", "PIP_FIND_LINKS", "PIP_NO_INDEX"): |
| 188 | + environment.pop(variable, None) |
| 189 | + result = subprocess.run(command, check=False, capture_output=True, text=True, env=environment) |
| 190 | + if result.returncode != 0: |
| 191 | + detail = (result.stderr or result.stdout).strip() |
| 192 | + raise ProjectSurfaceError(f"pip could not resolve {requirement}: {detail}") |
| 193 | + try: |
| 194 | + return json.loads(report_path.read_text(encoding="utf-8")) |
| 195 | + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: |
| 196 | + raise ProjectSurfaceError("pip did not produce a valid JSON resolution report") from error |
| 197 | + |
| 198 | + |
| 199 | +def _request_json(url: str) -> object: |
| 200 | + request = urllib.request.Request( |
| 201 | + url, |
| 202 | + headers={"Accept": "application/json", "User-Agent": "durable-workflow-pypi-project-surface-audit"}, |
| 203 | + ) |
| 204 | + with urllib.request.urlopen(request, timeout=30) as response: |
| 205 | + if response.status != 200: |
| 206 | + raise ProjectSurfaceError(f"{url} returned HTTP {response.status}") |
| 207 | + try: |
| 208 | + return json.loads(response.read().decode("utf-8")) |
| 209 | + except (UnicodeDecodeError, json.JSONDecodeError) as error: |
| 210 | + raise ProjectSurfaceError("authoritative PyPI project-root JSON is invalid") from error |
| 211 | + |
| 212 | + |
| 213 | +def main(argv: list[str] | None = None) -> int: |
| 214 | + parser = argparse.ArgumentParser(description=__doc__) |
| 215 | + parser.add_argument("--source-ref", required=True, help="Exact current prerelease source commit or tag") |
| 216 | + parser.add_argument("--attempts", type=int, default=1, help="PyPI root-metadata propagation attempts") |
| 217 | + parser.add_argument("--interval-seconds", type=float, default=0.0, help="Delay between metadata attempts") |
| 218 | + args = parser.parse_args(argv) |
| 219 | + if args.attempts < 1 or args.interval_seconds < 0: |
| 220 | + raise ProjectSurfaceError("attempts must be positive and interval-seconds must be non-negative") |
| 221 | + |
| 222 | + try: |
| 223 | + source = load_source_metadata(args.source_ref) |
| 224 | + except ReleaseMetadataError as error: |
| 225 | + raise ProjectSurfaceError(str(error)) from error |
| 226 | + project_json_url = f"https://pypi.org/pypi/{source.name}/json" |
| 227 | + last_error: BaseException | None = None |
| 228 | + legacy_version: str | None = None |
| 229 | + for attempt in range(1, args.attempts + 1): |
| 230 | + try: |
| 231 | + legacy_version = verify_project_json(_request_json(project_json_url), source) |
| 232 | + break |
| 233 | + except (ProjectSurfaceError, urllib.error.URLError) as error: |
| 234 | + last_error = error |
| 235 | + if attempt < args.attempts: |
| 236 | + time.sleep(args.interval_seconds) |
| 237 | + if legacy_version is None: |
| 238 | + assert last_error is not None |
| 239 | + raise ProjectSurfaceError( |
| 240 | + f"PyPI project-root metadata did not converge after {args.attempts} attempt(s): {last_error}" |
| 241 | + ) |
| 242 | + |
| 243 | + verify_pip_report(_pip_report(source.name, prerelease=False), source, source.registry_version) |
| 244 | + requirement = supported_prerelease_requirement(source) |
| 245 | + verify_pip_report(_pip_report(requirement, prerelease=True), source, source.registry_version) |
| 246 | + legacy_requirement = f"{source.name}=={legacy_version}" |
| 247 | + verify_pip_report(_pip_report(legacy_requirement, prerelease=False), source, legacy_version) |
| 248 | + print( |
| 249 | + f"PyPI project root, default install, and supported prerelease install select " |
| 250 | + f"{source.name} {source.registry_version}; " |
| 251 | + f"exact historical install {legacy_requirement} remains resolvable" |
| 252 | + ) |
| 253 | + return 0 |
| 254 | + |
| 255 | + |
| 256 | +if __name__ == "__main__": |
| 257 | + raise SystemExit(main()) |
0 commit comments