Scope JSON vulnerabilities to the scanned repository version

Filter OSV reports by repository version so another index cannot show
vulns until it is scanned. Move the Warehouse trim helper out of
Django-backed utils so unit tests collect without loading apps.

Assisted By: Cursor Grok 4.6

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: gerrod3

pulp_python/app/osv.py

@@ -0,0 +1,42 @@
+from packaging.version import InvalidVersion, Version
+
+
+def _osv_fixed_in(vuln):
+    """Extract PEP 440 fixed versions from an OSV vulnerability record."""
+    fixed = []
+    seen = set()
+    for affected in vuln.get("affected") or []:
+        for range_ in affected.get("ranges") or []:
+            for event in range_.get("events") or []:
+                if "fixed" not in event:
+                    continue
+                version = event["fixed"]
+                if version in seen:
+                    continue
+                try:
+                    Version(version)
+                except InvalidVersion:
+                    continue
+                seen.add(version)
+                fixed.append(version)
+    return fixed
+
+
+def osv_to_pypi_vulnerabilities(vulns):
+    """Trim OSV vulnerability records to the Warehouse JSON API shape."""
+    seen = {}
+    for vuln in vulns or []:
+        vuln_id = vuln.get("id")
+        if not vuln_id or vuln_id in seen:
+            continue
+        seen[vuln_id] = {
+            "id": vuln_id,
+            "source": "osv",
+            "link": f"https://osv.dev/vulnerability/{vuln_id}",
+            "aliases": vuln.get("aliases") or [],
+            "details": vuln.get("details"),
+            "summary": vuln.get("summary"),
+            "fixed_in": _osv_fixed_in(vuln),
+            "withdrawn": vuln.get("withdrawn"),
+        }
+    return list(seen.values())

pulp_python/app/utils.py

@@ -17,13 +17,15 @@
 from jinja2 import Template
 from packaging.requirements import Requirement
 from packaging.utils import canonicalize_name
-from packaging.version import InvalidVersion, Version, parse
+from packaging.version import InvalidVersion, parse
 from pypi_simple import ACCEPT_JSON_PREFERRED, ProjectPage
 
 from pulpcore.plugin.exceptions import TimeoutException
 from pulpcore.plugin.models import Artifact, Remote, VulnerabilityReport
 from pulpcore.plugin.util import get_domain
 
+from pulp_python.app.osv import osv_to_pypi_vulnerabilities
+
 log = logging.getLogger(__name__)
 
 
@@ -408,56 +410,20 @@ def python_content_to_json(
     full_metadata["info"] = python_content_to_info(latest_content[0])
     full_metadata["releases"] = python_content_to_releases(all_content, base_path, domain)
     full_metadata["urls"] = python_content_to_urls(latest_content, base_path, domain)
-    full_metadata["vulnerabilities"] = _vulnerabilities_for_content(latest_content)
+    full_metadata["vulnerabilities"] = _vulnerabilities_for_content(
+        latest_content, repository_version
+    )
     return full_metadata
 
 
-def _osv_fixed_in(vuln):
-    """Extract PEP 440 fixed versions from an OSV vulnerability record."""
-    fixed = []
-    seen = set()
-    for affected in vuln.get("affected") or []:
-        for range_ in affected.get("ranges") or []:
-            for event in range_.get("events") or []:
-                if "fixed" not in event:
-                    continue
-                version = event["fixed"]
-                if version in seen:
-                    continue
-                try:
-                    Version(version)
-                except InvalidVersion:
-                    continue
-                seen.add(version)
-                fixed.append(version)
-    return fixed
-
-
-def osv_to_pypi_vulnerabilities(vulns):
-    """Trim OSV vulnerability records to the Warehouse JSON API shape."""
-    seen = {}
-    for vuln in vulns or []:
-        vuln_id = vuln.get("id")
-        if not vuln_id or vuln_id in seen:
-            continue
-        seen[vuln_id] = {
-            "id": vuln_id,
-            "source": "osv",
-            "link": f"https://osv.dev/vulnerability/{vuln_id}",
-            "aliases": vuln.get("aliases") or [],
-            "details": vuln.get("details"),
-            "summary": vuln.get("summary"),
-            "fixed_in": _osv_fixed_in(vuln),
-            "withdrawn": vuln.get("withdrawn"),
-        }
-    return list(seen.values())
-
-
-def _vulnerabilities_for_content(contents):
-    """Load VulnerabilityReports for content units and trim them to Warehouse shape."""
-    if not contents:
+def _vulnerabilities_for_content(contents, repository_version=None):
+    """Load VulnerabilityReports scanned for this repository version and trim to Warehouse shape."""
+    if not contents or repository_version is None:
         return []
-    reports = VulnerabilityReport.objects.filter(content_id__in=[c.pk for c in contents])
+    reports = VulnerabilityReport.objects.filter(
+        content_id__in=[c.pk for c in contents],
+        repo_versions=repository_version,
+    )
     merged = []
     for vulns in reports.values_list("vulns", flat=True):
         if vulns:

pulp_python/tests/functional/api/test_pypi_json_vulnerabilities.py

@@ -59,6 +59,7 @@ def test_pypi_json_vulnerabilities_from_manual_scan(
     python_bindings,
     python_remote_factory,
     python_repo,
+    python_repo_factory,
     python_distribution_factory,
     monitor_task,
 ):
@@ -94,11 +95,26 @@ def test_pypi_json_vulnerabilities_from_manual_scan(
         repository_version=repo.latest_version_href,
     )
     assert packages.count >= 2
+    hrefs = []
     for content in packages.results:
         assert content.vuln_report is not None
         report = pulpcore_bindings.VulnReportApi.read(content.vuln_report)
         assert report.vulns
         assert "affected" in report.vulns[0]
+        hrefs.append(content.pulp_href)
+
+    other = python_repo_factory()
+    monitor_task(
+        python_bindings.RepositoriesPythonApi.modify(
+            other.pulp_href, {"add_content_units": hrefs}
+        ).task
+    )
+    other = python_bindings.RepositoriesPythonApi.read(other.pulp_href)
+    other_distro = python_distribution_factory(repository=other)
+    other_json = requests.get(
+        urljoin(_index_url(other_distro, bindings_cfg), f"pypi/{name}/json")
+    ).json()
+    assert other_json["vulnerabilities"] == []
 
 
 @pytest.mark.parallel

pulp_python/tests/unit/test_vulnerabilities.py

@@ -1,4 +1,4 @@
-from pulp_python.app.utils import osv_to_pypi_vulnerabilities
+from pulp_python.app.osv import osv_to_pypi_vulnerabilities
 
 
 def test_osv_to_pypi_maps_fixed_versions_and_skips_git_shas():