PyPI: Adjust JSON API use to handle upstream changes

CHANGELOG.md

@@ -8,6 +8,12 @@ All versions prior to 0.0.9 are untracked.
 
 ## [Unreleased]
 
+### Fixed
+
+* Fixed a breakage in hash-checking mode caused by a
+  [change to the PyPI JSON API](https://discuss.python.org/t/backwards-incompatible-change-to-pypi-json-api/17154)
+  ([#318](https://github.com/trailofbits/pip-audit/pull/318))
+
 ## [2.4.0]
 
 ### Added

pip_audit/_service/pypi.py

@@ -8,7 +8,6 @@
 from typing import List, Optional, Tuple, cast
 
 import requests
-from packaging.utils import canonicalize_version
 from packaging.version import InvalidVersion, Version
 
 from pip_audit._cache import caching_session
@@ -87,25 +86,10 @@ def query(self, spec: Dependency) -> Tuple[Dependency, List[VulnerabilityResult]
 
         # If the dependency has a hash explicitly listed, check it against the PyPI data
         if spec.hashes:
-            # NOTE: PyPI has lots of "legacy" version formats in old releases.
-            # To handle these, we attempt to parse and canonicalize them as
-            # PEP 440 versions, falling back on the unparsed version.
-            releases = {}
-            for r, v in response_json["releases"].items():
-                try:
-                    releases[canonicalize_version(Version(r))] = v
-                except InvalidVersion:
-                    releases[r] = v
-            release = releases.get(canonicalize_version(spec.version))
-            if release is None:
-                raise ServiceError(
-                    "Could not find release to compare hashes: "
-                    f"{spec.canonical_name} ({spec.version})"
-                )
             for hash_type, hash_values in spec.hashes.items():
                 for hash_value in hash_values:
                     found = False
-                    for dist in release:
+                    for dist in response_json["urls"]:
                         digests = dist["digests"]
                         pypi_hash = digests.get(hash_type)
                         if pypi_hash is not None and pypi_hash == hash_value:

test/service/test_pypi.py

@@ -274,7 +274,7 @@ def raise_for_status(self):
 
             def json(self):
                 return {
-                    "releases": {},
+                    "urls": [],
                     "vulnerabilities": [
                         {
                             "id": "VULN-0",
@@ -294,67 +294,3 @@ def json(self):
     dep = service.ResolvedDependency("foo", Version("1.0"), hashes={"sha256": ["package-hash"]})
     with pytest.raises(service.ServiceError):
         dict(pypi.query_all(iter([dep])))
-
-
-def test_pypi_legacy_release_versions_do_not_crash(cache_dir, monkeypatch):
-    def get_mock_response():
-        class MockResponse:
-            def raise_for_status(self):
-                pass
-
-            def json(self):
-                return {
-                    "releases": {
-                        # Sampled from real invalid ("legacy") versions on PyPI
-                        "0.8.0-final0": [],
-                        "0.1-bulbasaur": [],
-                        "0.8d": [],
-                        "2004d": [],
-                        "0.5.2.5.g5b3e942": [],
-                        "git-unknown": [],
-                        "0.7.1.fix1": [],
-                        "r1": [],
-                        "0.1.0.1.g46ab6a7": [],
-                        "1.8.-1": [],
-                        "9bbb4f3": [],
-                        "2.0.0-final": [],
-                        "2013-02-16": [],
-                        "1.0beta5prerelease": [],
-                        "1.1.0-2-g555364d-dirty": [],
-                        "0.2-test1": [],
-                        "1.2.dev-r25": [],
-                        # Some valid PEP 440 versions for good measure
-                        "0.8.0rc1": [],
-                        "0.9.0": [],
-                        "1.0.0": [{"digests": {"sha256": "fakesha"}}],
-                    },
-                    "vulnerabilities": [
-                        {
-                            "id": "VULN-0",
-                            "details": "The first vulnerability",
-                            "fixed_in": ["1.1"],
-                            "aliases": [],
-                        }
-                    ],
-                }
-
-        return MockResponse()
-
-    monkeypatch.setattr(
-        service.pypi, "caching_session", lambda _: get_mock_session(get_mock_response)
-    )
-
-    pypi = service.PyPIService(cache_dir)
-    dep = service.ResolvedDependency("foo", Version("1.0"), hashes={"sha256": ["fakesha"]})
-    results: Dict[service.Dependency, List[service.VulnerabilityResult]] = dict(
-        pypi.query_all(iter([dep]))
-    )
-    assert len(results) == 1
-    assert dep in results
-    assert len(results[dep]) == 1
-    assert results[dep][0] == service.VulnerabilityResult(
-        id="VULN-0",
-        description="The first vulnerability",
-        fix_versions=[Version("1.1")],
-        aliases=set(),
-    )