service/osv: guard against schema changes

CHANGELOG.md

@@ -18,6 +18,9 @@ All versions prior to 0.0.9 are untracked.
   connection failures to vulnerability sources was improved
   ([#287](https://github.com/trailofbits/pip-audit/pull/287))
 
+* Vulnerability sources: the OSV service is now more resilient
+  to schema changes ([#288](https://github.com/trailofbits/pip-audit/pull/288))
+
 ### Fixed
 
 * Vulnerability sources: a bug stemming from an incorrect assumption

pip_audit/_service/osv.py

@@ -79,6 +79,15 @@ def query(self, spec: Dependency) -> Tuple[Dependency, List[VulnerabilityResult]
             return spec, results
 
         for vuln in response_json["vulns"]:
+            # Sanity check: only the v1 schema is specified at the moment,
+            # and the code below probably won't work with future incompatible
+            # schemas without additional changes.
+            # The absence of a schema is treated as 1.0.0, per the OSV spec.
+            schema_version = Version(vuln.get("schema_version", "1.0.0"))
+            if schema_version.major != 1:
+                logger.warning(f"Unsupported OSV schema version: {schema_version}")
+                continue
+
             id = vuln["id"]
 
             # The summary is intended to be shorter, so we prefer it over

test/service/test_osv.py

@@ -138,6 +138,35 @@ def test_osv_skipped_dep():
     assert len(vulns) == 0
 
 
+@pytest.mark.parametrize("version", ["0.0.0", "2.0.0", "2.3.4"])
+def test_osv_unsupported_schema_version(monkeypatch, version):
+    logger = pretend.stub(warning=pretend.call_recorder(lambda s: None))
+    monkeypatch.setattr(service.osv, "logger", logger)
+
+    payload = {
+        "vulns": [
+            {"schema_version": version},
+        ]
+    }
+
+    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 logger.warning.calls == [pretend.call(f"Unsupported OSV schema version: {version}")]
+
+    assert len(results) == 1
+    assert dep in results
+
+    vulns = results[dep]
+    assert len(vulns) == 0
+
+
 @pytest.mark.parametrize(
     ["summary", "details", "description"],
     [