diff --git a/secator/kev.py b/secator/kev.py new file mode 100644 index 000000000..9ddc3f958 --- /dev/null +++ b/secator/kev.py @@ -0,0 +1,55 @@ +"""CISA Known Exploited Vulnerabilities (KEV) catalog helpers. + +The CISA KEV catalog (https://www.cisa.gov/known-exploited-vulnerabilities-catalog) +lists CVEs that are known to be actively exploited in the wild. Secator downloads the +feed once (cached to the data directory, like wordlists / payloads) and exposes the set +of KEV CVE IDs so that any emitted vulnerability whose CVE is known-exploited can be +tagged with ``kev``. +""" +import json + +from secator.config import CONFIG, download_file + +KEV_URL = 'https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json' + +# Lazily-loaded cache of upper-cased KEV CVE IDs. ``None`` means "not loaded yet"; an empty +# set means "loaded but unavailable" (offline / download failed) so we never retry per-vuln. +_KEV_CVE_IDS = None + + +def get_kev_cve_ids(): + """Return the set of KEV CVE IDs (upper-cased), downloading the feed once if needed. + + The result is memoized for the lifetime of the process. On any failure (offline mode, + download error, malformed feed) an empty set is cached and returned, so tagging simply + becomes a no-op instead of raising. + + Returns: + set[str]: Upper-cased CVE IDs present in the CISA KEV catalog. + """ + global _KEV_CVE_IDS + if _KEV_CVE_IDS is None: + _KEV_CVE_IDS = _load_kev_cve_ids() + return _KEV_CVE_IDS + + +def _load_kev_cve_ids(): + """Download and parse the CISA KEV feed, returning the set of CVE IDs.""" + from secator.utils import debug + try: + path = download_file( + KEV_URL, CONFIG.dirs.data, CONFIG.offline_mode, 'kev', + name='known_exploited_vulnerabilities.json' + ) + if not path: + return set() + with open(path) as f: + data = json.load(f) + return { + vuln['cveId'].upper() + for vuln in data.get('vulnerabilities', []) + if vuln.get('cveId') + } + except Exception as e: + debug(f'Failed to load CISA KEV catalog: {e}', sub='cve') + return set() diff --git a/secator/output_types/vulnerability.py b/secator/output_types/vulnerability.py index b726958f4..a2ac29dc4 100644 --- a/secator/output_types/vulnerability.py +++ b/secator/output_types/vulnerability.py @@ -68,6 +68,15 @@ def __post_init__(self): self.severity = self.severity.lower() # normalize severity self.severity_nb = severity_map.get(self.severity, 6) self.confidence_nb = severity_map[self.confidence] + self.tag_kev() + + def tag_kev(self): + """Add the ``kev`` tag when this vuln's CVE is in the CISA KEV catalog.""" + if not self.id or 'kev' in self.tags: + return + from secator.kev import get_kev_cve_ids + if self.id.upper() in get_kev_cve_ids(): + self.tags.append('kev') def __rich__(self): data = self.extra_data diff --git a/tests/unit/test_kev.py b/tests/unit/test_kev.py new file mode 100644 index 000000000..6a3aa4e25 --- /dev/null +++ b/tests/unit/test_kev.py @@ -0,0 +1,50 @@ +import unittest +from unittest.mock import patch + +import secator.kev as kev +from secator.output_types import Vulnerability + + +class TestKevTagging(unittest.TestCase): + def setUp(self): + # Reset the process-wide cache before each test so patches take effect. + kev._KEV_CVE_IDS = None + + def tearDown(self): + kev._KEV_CVE_IDS = None + + @patch('secator.kev._load_kev_cve_ids', return_value={'CVE-2021-44228'}) + def test_kev_tag_added_when_cve_in_catalog(self, _): + vuln = Vulnerability(name='Log4Shell', id='CVE-2021-44228') + self.assertIn('kev', vuln.tags) + + @patch('secator.kev._load_kev_cve_ids', return_value={'CVE-2021-44228'}) + def test_kev_tag_case_insensitive(self, _): + vuln = Vulnerability(name='Log4Shell', id='cve-2021-44228') + self.assertIn('kev', vuln.tags) + + @patch('secator.kev._load_kev_cve_ids', return_value={'CVE-2021-44228'}) + def test_kev_tag_not_added_when_cve_absent(self, _): + vuln = Vulnerability(name='Some vuln', id='CVE-2000-0001') + self.assertNotIn('kev', vuln.tags) + + @patch('secator.kev._load_kev_cve_ids', return_value={'CVE-2021-44228'}) + def test_kev_tag_not_duplicated(self, _): + vuln = Vulnerability(name='Log4Shell', id='CVE-2021-44228', tags=['kev']) + self.assertEqual(vuln.tags.count('kev'), 1) + + @patch('secator.kev._load_kev_cve_ids', return_value=set()) + def test_no_tag_when_catalog_unavailable(self, _): + vuln = Vulnerability(name='Log4Shell', id='CVE-2021-44228') + self.assertNotIn('kev', vuln.tags) + + def test_no_id_is_noop(self): + # No CVE id => no network / catalog access at all. + with patch('secator.kev._load_kev_cve_ids') as mock_load: + vuln = Vulnerability(name='Generic finding') + self.assertNotIn('kev', vuln.tags) + mock_load.assert_not_called() + + +if __name__ == '__main__': + unittest.main()