diff --git a/pyvulnerabilitylookup/__init__.py b/pyvulnerabilitylookup/__init__.py index df886a0..0aaa052 100644 --- a/pyvulnerabilitylookup/__init__.py +++ b/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: diff --git a/pyvulnerabilitylookup/api.py b/pyvulnerabilitylookup/api.py index 6bc15ae..31f7cb7 100644 --- a/pyvulnerabilitylookup/api.py +++ b/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]: diff --git a/tests/test_web.py b/tests/test_web.py index 2ada1d3..e30f79d 100644 --- a/tests/test_web.py +++ b/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: