Query the GCVE BCP-07 KEV catalogs; deprecate the CISA KEV helpers

pyvulnerabilitylookup/__init__.py

@@ -6,9 +6,9 @@
 
 from typing import Any
 
-from .api import PyVulnerabilityLookup
+from .api import CISA_KEV_ORIGIN, PyVulnerabilityLookup
 
-__all__ = ['PyVulnerabilityLookup']
+__all__ = ['CISA_KEV_ORIGIN', 'PyVulnerabilityLookup']
 
 
 def main() -> None:

pyvulnerabilitylookup/api.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import warnings
 
 from datetime import date, datetime, timedelta
 from importlib.metadata import version
@@ -26,6 +27,10 @@ def enable_full_debug() -> None:
     requests_log.propagate = True
 
 
+# Origin UUID of the CISA KEV catalog in the GCVE references (``gcve references --list``,
+# key ``kev``). Catalog UUIDs are global, so it is the same on every instance.
+CISA_KEV_ORIGIN = '405284c2-e461-4670-8979-7fd2c9755a60'
+
 class PyVulnerabilityLookup():
 
     def __init__(self, root_url: str='https://vulnerability.circl.lu', useragent: str | None=None, token: str | None=None,
@@ -977,34 +982,123 @@ def get_emb3d(self, emb3d_id: str) -> dict[str, Any]:
         r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'emb3d', emb3d_id))))
         return r.json()
 
-    # #### CISA KEV ####
+    # #### KEV (GCVE BCP-07 catalogs) ####
 
-    def get_cisa_kevs(self, *, page: int | None=None, per_page: int | None=None) -> dict[str, Any]:
-        '''Get the list of all CISA KEV
+    def get_kevs(self, /, *, page: int | None=None, per_page: int | None=None,
+                 vuln_id: str | None=None, origin: str | None=None,
+                 status_reason: str | None=None, exploited: str | bool | None=None,
+                 date_from: date | datetime | None=None,
+                 date_to: date | datetime | None=None,
+                 author: str | None=None) -> dict[str, Any]:
+        '''Get the Known Exploited Vulnerabilities (KEV) assertions, in the GCVE BCP-07 format.
+        The response is paginated.
+
+        A Vulnerability-Lookup instance holds one KEV catalog per origin (its own, and the
+        ones it imports or synchronizes: CISA KEV, ENISA, ...). The origin UUIDs are listed
+        on the instance's ``/kev-catalogs`` page and in the GCVE references.
 
         :param page: The page to get (default: 1)
-        :param per_page: The number of results per page (default: 100)
+        :param per_page: The number of results per page (default: 100, max: 1000)
+        :param vuln_id: The vulnerability ID to get the assertions of, across catalogs
+        :param origin: The UUID of the catalog (origin instance) to list, e.g. ``CISA_KEV_ORIGIN``
+        :param status_reason: One of: 'confirmed', 'suspected', 'disputed', 'historical', 'unknown', 'withdrawn'
+        :param exploited: ``True`` (default, the current KEV set), ``False`` (withdrawn assertions only) or ``'all'``
+        :param date_from: The date from which the assertions were recorded
+        :param date_to: The date to which the assertions were recorded
+        :param author: The login of the author of the assertions
         '''
         params: dict[str, Any] = {}
         if page is not None:
             params['page'] = page
         if per_page is not None:
             params['per_page'] = per_page
-        r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'cisa_kev'))), params=params)
+        if vuln_id:
+            params['vuln_id'] = vuln_id
+        if origin:
+            params['vulnerability_lookup_origin'] = origin
+        if status_reason:
+            params['status_reason'] = status_reason
+        if exploited is not None:
+            params['exploited'] = exploited if isinstance(exploited, str) else str(exploited).lower()
+        if date_from:
+            if isinstance(date_from, datetime):
+                date_from = date_from.date()
+            params['date_from'] = date_from.isoformat()
+        if date_to:
+            if isinstance(date_to, datetime):
+                date_to = date_to.date()
+            params['date_to'] = date_to.isoformat()
+        if author:
+            params['author'] = author
+        r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'kev'))), params=params)
         return r.json()
 
-    def get_cisa_kevs_iter(self) -> Generator[dict[str, Any]]:
-        '''Iterate over the CISA KEV list.
+    def get_kevs_iter(self, /, *, vuln_id: str | None=None, origin: str | None=None,
+                      status_reason: str | None=None, exploited: str | bool | None=None,
+                      date_from: date | datetime | None=None,
+                      date_to: date | datetime | None=None,
+                      author: str | None=None) -> Generator[dict[str, Any]]:
+        '''Iterate over the KEV assertions (GCVE BCP-07 format).
+
+        :param vuln_id: The vulnerability ID to get the assertions of, across catalogs
+        :param origin: The UUID of the catalog (origin instance) to list, e.g. ``CISA_KEV_ORIGIN``
+        :param status_reason: One of: 'confirmed', 'suspected', 'disputed', 'historical', 'unknown', 'withdrawn'
+        :param exploited: ``True`` (default, the current KEV set), ``False`` (withdrawn assertions only) or ``'all'``
+        :param date_from: The date from which the assertions were recorded
+        :param date_to: The date to which the assertions were recorded
+        :param author: The login of the author of the assertions
         '''
         page = 1
-        per_page = 20
+        per_page = 1000
         while True:
-            r = self.get_cisa_kevs(page=page, per_page=per_page)
+            r = self.get_kevs(page=page, per_page=per_page, vuln_id=vuln_id, origin=origin,
+                              status_reason=status_reason, exploited=exploited,
+                              date_from=date_from, date_to=date_to, author=author)
             if not r['data']:
                 break
             yield from r['data']
             page += 1
 
+    def get_kev(self, kev_uuid: str) -> dict[str, Any]:
+        '''Get a KEV assertion by its UUID.
+
+        :param kev_uuid: The UUID of the assertion
+        '''
+        r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'kev', kev_uuid))))
+        return r.json()
+
+    def get_kev_by_origin(self, origin: str, vuln_id: str) -> dict[str, Any]:
+        '''Get the assertion a KEV catalog holds for a vulnerability.
+
+        :param origin: The UUID of the catalog (origin instance), e.g. ``CISA_KEV_ORIGIN``
+        :param vuln_id: The vulnerability ID
+        '''
+        r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'kev', 'origin', origin, vuln_id))))
+        return r.json()
+
+    def get_cisa_kevs(self, *, page: int | None=None, per_page: int | None=None) -> dict[str, Any]:
+        '''Get the CISA KEV catalog. Deprecated: use ``get_kevs(origin=CISA_KEV_ORIGIN)``.
+
+        The ``/api/cisa_kev/`` endpoint this method used to call was removed from
+        Vulnerability-Lookup (a non-BCP-07 mirror of the CISA list, retired in favour of
+        the BCP-07 catalog). This now returns the BCP-07 assertions of the CISA KEV
+        catalog, whose entries are shaped differently from the former CISA records.
+
+        :param page: The page to get (default: 1)
+        :param per_page: The number of results per page (default: 100)
+        '''
+        warnings.warn('get_cisa_kevs is deprecated, use get_kevs(origin=CISA_KEV_ORIGIN); '
+                      'the entries are now GCVE BCP-07 assertions.', DeprecationWarning, stacklevel=2)
+        return self.get_kevs(page=page, per_page=per_page, origin=CISA_KEV_ORIGIN)
+
+    def get_cisa_kevs_iter(self) -> Generator[dict[str, Any]]:
+        '''Iterate over the CISA KEV catalog. Deprecated: use ``get_kevs_iter(origin=CISA_KEV_ORIGIN)``.
+        See ``get_cisa_kevs``.
+        '''
+        warnings.warn('get_cisa_kevs_iter is deprecated, use get_kevs_iter(origin=CISA_KEV_ORIGIN); '
+                      'the entries are now GCVE BCP-07 assertions.', DeprecationWarning, stacklevel=2)
+        yield from self.get_kevs_iter(origin=CISA_KEV_ORIGIN)
+
     # #### GCVE ####
 
     def get_gcves(self, short_name: str | None=None, *, page: int | None=None, per_page: int | None=None) -> dict[str, Any]:

tests/test_web.py

@@ -7,7 +7,7 @@
 
 from datetime import datetime, timezone, timedelta, date
 
-from pyvulnerabilitylookup import PyVulnerabilityLookup
+from pyvulnerabilitylookup import CISA_KEV_ORIGIN, PyVulnerabilityLookup
 
 # NOTE:
 # * to run the tests with a pre-configured admin key:
@@ -491,12 +491,13 @@ def test_product(self) -> None:
         products = [p for p in self.client.get_products_iter(name='misp')]
         self.assertTrue(len(products) > 0)
         self.assertTrue(products[0]['name'] == 'MISP', products)
+        # the organization has several products; the listing is not ordered by name
         products = [p for p in self.client.get_products_iter(organization_name='misp')]
         self.assertTrue(len(products) > 0)
-        self.assertTrue(products[0]['name'] == 'MISP', products)
+        self.assertTrue(any(p['name'] == 'MISP' for p in products), products)
         products = [p for p in self.client.get_products_iter(organization_uuid='e8149169-7327-4930-a576-afd4f76f2e61')]
         self.assertTrue(len(products) > 0)
-        self.assertTrue(products[0]['name'] == 'MISP', products)
+        self.assertTrue(any(p['name'] == 'MISP' for p in products), products)
 
     def test_emb3d(self) -> None:
         if not self.public_test:
@@ -510,12 +511,26 @@ def test_emb3d(self) -> None:
         all_cves = set(emb3d_details['CVE'][0].keys())
         self.assertTrue('CVE-2022-1118' in all_cves, emb3d_details)
 
-    def test_cisa_kev(self) -> None:
+    def test_kev(self) -> None:
         if not self.public_test:
             return None
-        # test iterator
-        for kev in self.client.get_cisa_kevs_iter():
-            self.assertTrue('cveID' in kev, kev)
+        # test iterator, restricted to the CISA KEV catalog (GCVE BCP-07 assertions)
+        kevs = []
+        for kev in self.client.get_kevs_iter(origin=CISA_KEV_ORIGIN):
+            self.assertEqual(kev['gcve']['origin_uuid'], CISA_KEV_ORIGIN, kev)
+            self.assertTrue(kev['status']['exploited'], kev)
+            kevs.append(kev)
+        self.assertTrue(kevs)
+        first = kevs[0]
+        vuln_id = first['vulnerability']['vulnId']
+        self.assertEqual(self.client.get_kev(first['uuid'])['uuid'], first['uuid'])
+        self.assertEqual(self.client.get_kev_by_origin(CISA_KEV_ORIGIN, vuln_id)['uuid'], first['uuid'])
+        # the same vulnerability across every catalog
+        across = self.client.get_kevs(vuln_id=vuln_id)
+        self.assertTrue(any(k['uuid'] == first['uuid'] for k in across['data']), across)
+        # the CISA helpers are deprecated but still answer, with the BCP-07 shape
+        with self.assertWarns(DeprecationWarning):
+            self.assertTrue('vulnerability' in next(self.client.get_cisa_kevs_iter()))
 
     def test_gcve(self) -> None:
         if not self.public_test: