osv: fix incorrect schema assumptions

CHANGELOG.md

@@ -14,6 +14,12 @@ All versions prior to 0.0.9 are untracked.
   faster and more responsive
   ([#283](https://github.com/trailofbits/pip-audit/pull/283))
 
+### Fixed
+
+* Vulnerability sources: a bug stemming from an incorrect assumption
+  about OSV's schema guarantees was fixed
+  ([#284](https://github.com/trailofbits/pip-audit/pull/284))
+
 ## [2.3.1] - 2022-05-24
 
 ### Fixed

pip_audit/_service/osv.py

@@ -3,6 +3,7 @@
 """
 
 import json
+import logging
 from pathlib import Path
 from typing import List, Optional, Tuple, cast
 
@@ -18,6 +19,8 @@
     VulnerabilityService,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class OsvService(VulnerabilityService):
     """
@@ -77,10 +80,26 @@ def query(self, spec: Dependency) -> Tuple[Dependency, List[VulnerabilityResult]
 
         for vuln in response_json["vulns"]:
             id = vuln["id"]
-            description = vuln["details"]
-            aliases = set(vuln["aliases"])
+
+            # The summary is intended to be shorter, so we prefer it over
+            # details, if present. However, neither is required.
+            description = vuln.get("summary")
+            if description is None:
+                description = vuln.get("details")
+            if description is None:
+                description = "N/A"
+
+            aliases = set(vuln.get("aliases", []))
+
+            # OSV doesn't mandate this field either. There's very little we
+            # can do without it, so we skip any results that are missing it.
+            affecteds = vuln.get("affected")
+            if affecteds is None:
+                logger.warning(f"OSV vuln entry '{id}' is missing 'affected' list")
+                continue
+
             fix_versions: List[Version] = []
-            for affected in vuln["affected"]:
+            for affected in affecteds:
                 pkg = affected["package"]
                 # We only care about PyPI versions
                 if pkg["name"] == spec.canonical_name and pkg["ecosystem"] == "PyPI":

test/service/test_osv.py

@@ -125,3 +125,80 @@ def test_osv_skipped_dep():
 
     vulns = results[dep]
     assert len(vulns) == 0
+
+
+@pytest.mark.parametrize(
+    ["summary", "details", "description"],
+    [
+        ("fakesummary", "fakedetails", "fakesummary"),
+        (None, "fakedetails", "fakedetails"),
+        (None, None, "N/A"),
+    ],
+)
+def test_osv_vuln_description_fallbacks(monkeypatch, summary, details, description):
+    payload = {
+        "vulns": [
+            {
+                "id": "fakeid",
+                "summary": summary,
+                "details": details,
+                "affected": [
+                    {
+                        "package": {"name": "foo", "ecosystem": "PyPI"},
+                        "ranges": [{"type": "ECOSYSTEM", "events": [{"fixed": "1.0.1"}]}],
+                    }
+                ],
+            }
+        ],
+    }
+
+    response = pretend.stub(raise_for_status=lambda: None, json=lambda: payload)
+    post = pretend.call_recorder(lambda *a, **kw: response)
+
+    osv = service.OsvService()
+    monkeypatch.setattr(osv.session, "post", post)
+
+    dep = service.ResolvedDependency("foo", Version("1.0.0"))
+    results = dict(osv.query_all(iter([dep])))
+
+    assert len(results) == 1
+    assert dep in results
+
+    vulns = results[dep]
+    assert len(vulns) == 1
+
+    assert vulns[0].description == description
+
+
+def test_osv_vuln_affected_missing(monkeypatch):
+    logger = pretend.stub(warning=pretend.call_recorder(lambda s: None))
+    monkeypatch.setattr(service.osv, "logger", logger)
+
+    payload = {
+        "vulns": [
+            {
+                "id": "fakeid",
+                "summary": "fakesummary",
+                "details": "fakedetails",
+            }
+        ],
+    }
+
+    response = pretend.stub(raise_for_status=lambda: None, json=lambda: payload)
+    post = pretend.call_recorder(lambda *a, **kw: response)
+
+    osv = service.OsvService()
+    monkeypatch.setattr(osv.session, "post", post)
+
+    dep = service.ResolvedDependency("foo", Version("1.0.0"))
+    results = dict(osv.query_all(iter([dep])))
+
+    assert len(results) == 1
+    assert dep in results
+
+    vulns = results[dep]
+    assert len(vulns) == 0
+
+    assert logger.warning.calls == [
+        pretend.call("OSV vuln entry 'fakeid' is missing 'affected' list")
+    ]