feat(output-types): tag known-exploited vulnerabilities with kev (#1319)

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()

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

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()