From f2398495714675403fc4113efaada5a255268479 Mon Sep 17 00:00:00 2001 From: Ryan Wilson Date: Thu, 27 Aug 2026 03:53:21 -0500 Subject: [PATCH] Add KEV/EPSS CVE enrichment and match confidence scoring Also fixes a false-positive bug where an unversioned fingerprint matched every bounded CVE for its product instead of only unbounded ones. --- .gitignore | 23 ---- bitprobe/bitprobe.py | 35 ++++- bitprobe/plugins/cve_correlation.py | 49 ++++++- bitprobe/scanner/cve_db_manager.py | 192 ++++++++++++++++++++++++++-- bitprobe/scanner/cve_matcher.py | 10 ++ tests/test_cve_db_manager.py | 189 +++++++++++++++++++++++++++ 6 files changed, 457 insertions(+), 41 deletions(-) diff --git a/.gitignore b/.gitignore index 98aa15a..e7b9f27 100644 --- a/.gitignore +++ b/.gitignore @@ -65,26 +65,3 @@ private/ bitprobe/.git/ bitprobe/data/cve_db.sqlite -# Spartan AI config (contains API keys) -.spartan/ai.env - -#Other -SKILL.md -AGENTS.md - -# Spartan local planning/memory state -- not for the public repo -.planning/ -.memory/ - -# Local OpenCode config -.opencode/ - -# Personal AI workflow tooling -- not project convention, not for the public repo -.claude/ - -# Isolated feature worktrees -.worktrees/ - -# Local-only files -CLAUDE.md -skills-lock.json diff --git a/bitprobe/bitprobe.py b/bitprobe/bitprobe.py index 7681dcb..d363bba 100755 --- a/bitprobe/bitprobe.py +++ b/bitprobe/bitprobe.py @@ -15,7 +15,7 @@ from scanner.engine import ScanEngine from scanner.config import ScanConfig, SCAN_PROFILES from scanner.asn_db_updater import update_asn_db -from scanner.cve_db_manager import update_cve_database, get_stats +from scanner.cve_db_manager import update_cve_database, update_kev_epss, get_stats from scanner.cve_db_bootstrap import update_with_snapshot_policy @@ -221,6 +221,16 @@ def main() -> int: action="store_true", help="Use direct NVD synchronization without downloading a snapshot", ) + cve_parser.add_argument( + "--kev-epss-only", + action="store_true", + help="Only refresh CISA KEV / FIRST EPSS enrichment, skip the NVD sync", + ) + cve_parser.add_argument( + "--skip-kev-epss", + action="store_true", + help="Skip CISA KEV / FIRST EPSS enrichment during this update", + ) cve_stats_parser = subparsers.add_parser( "cve-stats", @@ -247,9 +257,24 @@ def main() -> int: years = getattr(args, "years", None) days = getattr(args, "days", None) snapshot_only = getattr(args, "snapshot_only", False) + kev_epss_only = getattr(args, "kev_epss_only", False) + skip_kev_epss = getattr(args, "skip_kev_epss", False) direct = full_sync or years is not None or days is not None or getattr(args, "no_snapshot", False) if snapshot_only and direct: raise ValueError("--snapshot-only cannot be combined with direct-NVD options") + if kev_epss_only and skip_kev_epss: + raise ValueError("--kev-epss-only cannot be combined with --skip-kev-epss") + if kev_epss_only and (direct or snapshot_only): + raise ValueError("--kev-epss-only cannot be combined with NVD sync options") + + if kev_epss_only: + counts = update_kev_epss(verbose=verbose) + print( + f"[+] KEV/EPSS enrichment updated: " + f"{counts['kev_updated']} KEV, {counts['epss_updated']} EPSS" + ) + return 0 + if direct: count = update_cve_database( days=days if days is not None else 30, @@ -267,6 +292,12 @@ def main() -> int: if snapshot_only: print("[+] CVE database snapshot installed") return 0 + if not skip_kev_epss: + counts = update_kev_epss(verbose=verbose) + print( + f"[+] KEV/EPSS enrichment updated: " + f"{counts['kev_updated']} KEV, {counts['epss_updated']} EPSS" + ) print(f"[+] CVE database updated with {count} entries") return 0 except Exception as e: @@ -283,6 +314,8 @@ def main() -> int: print(f"Coverage: {stats.get('coverage_mode', 'unknown')}") print(f"NVD Cursor: {stats.get('nvd_cursor', 'Never')}") print(f"Last Updated: {stats.get('last_updated', 'Never')}") + print(f"CISA KEV flagged: {stats.get('kev_count', 0)}") + print(f"EPSS scored: {stats.get('epss_count', 0)}") print("\nBy Severity:") for sev, count in stats.get('severity_counts', {}).items(): print(f" {sev.upper()}: {count}") diff --git a/bitprobe/plugins/cve_correlation.py b/bitprobe/plugins/cve_correlation.py index abfe76b..5f31fe7 100755 --- a/bitprobe/plugins/cve_correlation.py +++ b/bitprobe/plugins/cve_correlation.py @@ -199,6 +199,40 @@ def _append_cve_finding( ) -> None: guidance = self._generate_contextual_guidance(tech_name, cve_id) refs = references or [] + confidence = match.get("confidence", "low") + kev = bool(match.get("kev")) + kev_date_added = match.get("kev_date_added") + epss_score = match.get("epss_score") + epss_percentile = match.get("epss_percentile") + + # CISA KEV means this CVE is confirmed exploited in the wild right + # now — that outweighs a CVSS-bucket severity computed in the + # abstract, so it overrides rather than just adding to it. + if kev: + severity = "critical" + + remediation = ( + f"Upgrade {tech_name} to a patched version. " + "See CVE details for specific fixed versions." + ) + if kev: + remediation = ( + "CISA's Known Exploited Vulnerabilities catalog lists this CVE as " + "actively exploited in the wild" + + (f" (added {kev_date_added})" if kev_date_added else "") + + f". Treat as critical and patch {tech_name} immediately, " + "regardless of match confidence below. " + ) + remediation + if confidence != "confirmed": + remediation += ( + " Confidence: low — this was matched on product name without a " + "version-range check (either no version was fingerprinted, or the " + "CVE record itself carries no version bound), so verify manually " + "before treating it as a confirmed finding. Note that vendor-patched " + "or distro-packaged builds can carry a backported fix while still " + "reporting an older upstream version string." + ) + findings.append( Finding( plugin_name=self.get_name(), @@ -212,12 +246,14 @@ def _append_cve_finding( "cve_id": cve_id, "cvss_score": cvss, "affected_versions": match.get("affected_versions", ""), + "confidence": confidence, + "kev": kev, + "kev_date_added": kev_date_added, + "epss_score": epss_score, + "epss_percentile": epss_percentile, "references": refs[:3], }, - remediation=( - f"Upgrade {tech_name} to a patched version. " - "See CVE details for specific fixed versions." - ), + remediation=remediation, attack_scenario=guidance["attack"], defense_strategy=guidance["defense"], mitigation_plan=guidance["mitigation"], @@ -292,6 +328,11 @@ def scan(self, url_info: Dict, request_handler) -> List[Finding]: "matched_product": tech_name, "detected_version": tech_version, "affected_versions": "see NVD advisory", + "confidence": cve_row.get("confidence", "low"), + "kev": cve_row.get("kev", False), + "kev_date_added": cve_row.get("kev_date_added"), + "epss_score": cve_row.get("epss_score"), + "epss_percentile": cve_row.get("epss_percentile"), } self._append_cve_finding( findings, diff --git a/bitprobe/scanner/cve_db_manager.py b/bitprobe/scanner/cve_db_manager.py index a0dd6d8..fad3004 100644 --- a/bitprobe/scanner/cve_db_manager.py +++ b/bitprobe/scanner/cve_db_manager.py @@ -5,6 +5,9 @@ Manages SQLite database for CVE tracking with NVD feed integration. """ +import csv +import gzip +import io import sqlite3 import json import os @@ -19,6 +22,8 @@ from scanner.update_state import get_state_timestamp, set_state_timestamp, merge_section NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0" +KEV_FEED_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" +EPSS_BULK_URL = "https://epss.empiricalsecurity.com/epss_scores-current.csv.gz" DEFAULT_STALE_DAYS = 7 # NVD rate limits (https://nvd.nist.gov/developers/start-here): 5 req/30s # without a key (6.0s spacing), 50 req/30s with one (0.6s spacing, +margin). @@ -409,6 +414,27 @@ def describe_cve_db_local_status() -> str: return f"ok ({count} CVEs loaded)" +def _ensure_enrichment_columns(cursor: sqlite3.Cursor) -> None: + """ + Add KEV/EPSS columns to cve_entries if they're missing. + + `CREATE TABLE IF NOT EXISTS` is a no-op against a database that was + built before these columns existed, so databases installed from an + older cve-db-* release snapshot need them added in place rather than + requiring a full rebuild. + """ + existing = {row[1] for row in cursor.execute("PRAGMA table_info(cve_entries)")} + additions = { + "kev": "BOOLEAN DEFAULT 0", + "kev_date_added": "TEXT", + "epss_score": "REAL", + "epss_percentile": "REAL", + } + for column, ddl in additions.items(): + if column not in existing: + cursor.execute(f"ALTER TABLE cve_entries ADD COLUMN {column} {ddl}") + + def init_cve_database(): """Initialize SQLite database with CVE schema.""" migrate_legacy_cve_database() @@ -427,10 +453,15 @@ def init_cve_database(): cvss_vector TEXT, published_date TEXT, last_modified TEXT, - "references" TEXT + "references" TEXT, + kev BOOLEAN DEFAULT 0, + kev_date_added TEXT, + epss_score REAL, + epss_percentile REAL ) """) - + _ensure_enrichment_columns(cursor) + # Product mappings for version matching cursor.execute(""" CREATE TABLE IF NOT EXISTS cve_products ( @@ -1046,6 +1077,109 @@ def update_cve_database( return count +def update_kev_data(verbose: bool = False) -> int: + """ + Flag CVEs already in the local database that CISA's Known Exploited + Vulnerabilities catalog lists as actively exploited in the wild. + + Only updates existing rows — a KEV entry for a CVE our local NVD data + doesn't have yet is skipped rather than inserted as a stub, since we + have no description/CVSS/CPE data for it. + """ + init_cve_database() + response = requests.get(KEV_FEED_URL, timeout=60) + response.raise_for_status() + vulnerabilities = response.json().get("vulnerabilities", []) + + rows = [ + (vuln.get("dateAdded"), vuln["cveID"]) + for vuln in vulnerabilities + if vuln.get("cveID") + ] + if verbose: + print(f"[VERBOSE] Parsed {len(rows)} entries from CISA KEV catalog") + + conn = _connect() + try: + cursor = conn.cursor() + # Clear stale flags first so a CVE CISA later removes from the + # catalog doesn't stay marked forever. + cursor.execute("UPDATE cve_entries SET kev = 0, kev_date_added = NULL WHERE kev = 1") + cursor.executemany( + "UPDATE cve_entries SET kev = 1, kev_date_added = ? WHERE cve_id = ?", + rows, + ) + updated = max(cursor.rowcount, 0) + conn.commit() + finally: + conn.close() + + print(f"[+] CISA KEV flag set for {updated} known CVEs") + return updated + + +def update_epss_data(verbose: bool = False) -> int: + """ + Refresh FIRST EPSS (Exploit Prediction Scoring System) scores for CVEs + already in the local database, from FIRST's daily bulk export. + + Only updates existing rows, same reasoning as update_kev_data: EPSS + scores for CVEs we don't have NVD data for yet aren't useful without + the rest of the record. + """ + init_cve_database() + response = requests.get(EPSS_BULK_URL, timeout=60) + response.raise_for_status() + raw = gzip.decompress(response.content).decode("utf-8") + + rows = [] + for row in csv.reader(io.StringIO(raw)): + if not row or row[0].startswith("#") or row[0] == "cve" or len(row) < 3: + continue + cve_id, score, percentile = row[0], row[1], row[2] + try: + rows.append((float(score), float(percentile), cve_id)) + except ValueError: + continue + if verbose: + print(f"[VERBOSE] Parsed {len(rows)} EPSS scores from bulk export") + + conn = _connect() + try: + cursor = conn.cursor() + cursor.executemany( + "UPDATE cve_entries SET epss_score = ?, epss_percentile = ? WHERE cve_id = ?", + rows, + ) + updated = max(cursor.rowcount, 0) + conn.commit() + finally: + conn.close() + + print(f"[+] EPSS score updated for {updated} known CVEs") + return updated + + +def update_kev_epss(verbose: bool = False) -> Dict[str, int]: + """ + Refresh both CISA KEV and FIRST EPSS enrichment. Each source is + independent of the other and of the core NVD sync, so a failure in + one (e.g. a feed being temporarily unreachable) doesn't block the + other or the primary CVE update. + """ + kev_updated = 0 + epss_updated = 0 + try: + kev_updated = update_kev_data(verbose=verbose) + except Exception as e: + print(f"[!] CISA KEV update failed: {e}") + try: + epss_updated = update_epss_data(verbose=verbose) + except Exception as e: + print(f"[!] FIRST EPSS update failed: {e}") + return {"kev_updated": kev_updated, "epss_updated": epss_updated} + + def _extract_cpe_matches_from_node(node: Dict) -> List[Dict]: """ Extract product entries from one NVD configuration node, recursing into @@ -1291,9 +1425,10 @@ def query_cves( # Map cpe_names to their expected vendor strings for filtering placeholders = ','.join('?' * len(cpe_names)) query = f""" - SELECT DISTINCT - c.cve_id, c.description, c.severity, + SELECT DISTINCT + c.cve_id, c.description, c.severity, c.cvss_score, c."references", c.published_date, + c.kev, c.kev_date_added, c.epss_score, c.epss_percentile, p.version_start, p.version_end, p.version_start_including, p.version_end_including FROM cve_entries c @@ -1317,12 +1452,13 @@ def query_cves( cves = {} for row in rows: cve_id = row['cve_id'] - if cve_id in cves: - # Already matched via a different product/version row for - # this CVE; each row is an independent vulnerable - # configuration, so one match is enough. - continue - if version and not version_in_range( + # version_in_range already handles version=None correctly (it + # only matches rows with no version bounds at all); previously + # this was guarded by `if version and ...`, which skipped the + # range check entirely when no version was detected and let an + # unversioned fingerprint match every bounded CVE for that + # product, not just the unbounded ones. + if not version_in_range( version, row['version_start'], row['version_end'], @@ -1330,15 +1466,37 @@ def query_cves( max_inclusive=bool(row['version_end_including']), ): continue + # "confirmed" means an actual detected version was checked + # against a real bounded range; "low" means either no version + # was detected, or the CVE record itself carries no version + # bound (so it was matched purely on product name). + confidence = ( + 'confirmed' + if version and (row['version_start'] or row['version_end']) + else 'low' + ) + if cve_id in cves: + # Already matched via a different product/version row for + # this CVE; each row is an independent vulnerable + # configuration, so one match is enough, but prefer to + # surface the more confident of the two if both occur. + if confidence == 'confirmed': + cves[cve_id]['confidence'] = 'confirmed' + continue cves[cve_id] = { 'cve_id': cve_id, 'description': row['description'], 'severity': row['severity'], 'cvss_score': row['cvss_score'], 'published_date': row['published_date'], - 'references': json.loads(row['references'] or '[]') + 'references': json.loads(row['references'] or '[]'), + 'confidence': confidence, + 'kev': bool(row['kev']), + 'kev_date_added': row['kev_date_added'], + 'epss_score': row['epss_score'], + 'epss_percentile': row['epss_percentile'], } - + return list(cves.values()) finally: @@ -1370,7 +1528,13 @@ def get_stats() -> Dict[str, Any]: cursor.execute("SELECT value FROM metadata WHERE key = 'last_updated'") last_updated = cursor.fetchone() metadata = read_cve_metadata(conn) - + + cursor.execute("SELECT COUNT(*) FROM cve_entries WHERE kev = 1") + kev_count = cursor.fetchone()[0] + + cursor.execute("SELECT COUNT(*) FROM cve_entries WHERE epss_score IS NOT NULL") + epss_count = cursor.fetchone()[0] + return { 'total_cves': total_cves, 'total_products': total_products, @@ -1380,6 +1544,8 @@ def get_stats() -> Dict[str, Any]: 'coverage_start': metadata.get('coverage_start'), 'coverage_end': metadata.get('coverage_end'), 'nvd_cursor': metadata.get('nvd_cursor'), + 'kev_count': kev_count, + 'epss_count': epss_count, } finally: diff --git a/bitprobe/scanner/cve_matcher.py b/bitprobe/scanner/cve_matcher.py index 27b0df3..341535d 100644 --- a/bitprobe/scanner/cve_matcher.py +++ b/bitprobe/scanner/cve_matcher.py @@ -279,10 +279,20 @@ def match_technology_to_cve(tech_name: str, tech_version: Optional[str], cve_ent min_inclusive=product.get("min_inclusive", True), max_inclusive=product.get("max_inclusive", True), ): + # "confirmed" means a detected version was checked against + # an actual bounded range; "low" means either no version + # was detected, or the CVE record has no version bound at + # all (matched on product name alone). + confidence = ( + "confirmed" + if tech_version and (product["min_version"] or product["max_version"]) + else "low" + ) return { "matched_product": product["product"], "detected_version": tech_version, "affected_versions": f"{product['min_version'] or 'any'} - {product['max_version'] or 'any'}", + "confidence": confidence, } return None diff --git a/tests/test_cve_db_manager.py b/tests/test_cve_db_manager.py index 00a31b4..efa6b34 100644 --- a/tests/test_cve_db_manager.py +++ b/tests/test_cve_db_manager.py @@ -9,6 +9,8 @@ if str(_BITPROBE) not in sys.path: sys.path.insert(0, str(_BITPROBE)) +import gzip + import scanner.cve_db_manager as cve_db_manager from scanner.cve_db_manager import ( NVD_SLEEP_NO_KEY, @@ -18,9 +20,25 @@ _nvd_inter_request_sleep, init_cve_database, query_cves, + update_epss_data, + update_kev_data, ) +class _FakeResponse: + def __init__(self, *, json_data=None, content: bytes = b"", status_code: int = 200) -> None: + self._json_data = json_data + self.content = content + self.status_code = status_code + + def json(self): + return self._json_data + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise cve_db_manager.requests.HTTPError(f"HTTP {self.status_code}") + + def _cpe_match(criteria: str, vulnerable: bool = True) -> dict: return {"vulnerable": vulnerable, "criteria": criteria} @@ -199,3 +217,174 @@ def test_query_cves_respects_exclusive_upper_bound(monkeypatch, tmp_path: Path) ) assert query_cves("apache", version="2.4.49") != [] assert query_cves("apache", version="2.4.50") == [] + + +def test_update_kev_data_flags_known_cve(monkeypatch, tmp_path: Path) -> None: + db_path = tmp_path / "cve.sqlite" + monkeypatch.setattr(cve_db_manager, "CVE_DB_PATH", str(db_path)) + init_cve_database() + conn = cve_db_manager._connect() + try: + conn.execute( + "INSERT INTO cve_entries (cve_id, description, severity, cvss_score, \"references\") " + "VALUES ('CVE-2024-00001', 'desc', 'high', 7.5, '[]')" + ) + conn.commit() + finally: + conn.close() + + kev_payload = { + "vulnerabilities": [ + {"cveID": "CVE-2024-00001", "dateAdded": "2024-02-01"}, + # A KEV entry for a CVE our local NVD data doesn't have yet + # should be skipped, not inserted as a stub row. + {"cveID": "CVE-2024-99999", "dateAdded": "2024-02-01"}, + ] + } + monkeypatch.setattr( + cve_db_manager.requests, + "get", + lambda url, timeout=60: _FakeResponse(json_data=kev_payload), + ) + + updated = update_kev_data() + assert updated == 1 + + conn = cve_db_manager._connect() + try: + row = conn.execute( + "SELECT kev, kev_date_added FROM cve_entries WHERE cve_id = 'CVE-2024-00001'" + ).fetchone() + finally: + conn.close() + assert row == (1, "2024-02-01") + + +def test_update_kev_data_clears_stale_flags(monkeypatch, tmp_path: Path) -> None: + db_path = tmp_path / "cve.sqlite" + monkeypatch.setattr(cve_db_manager, "CVE_DB_PATH", str(db_path)) + init_cve_database() + conn = cve_db_manager._connect() + try: + conn.execute( + "INSERT INTO cve_entries (cve_id, description, severity, cvss_score, \"references\", kev, kev_date_added) " + "VALUES ('CVE-2024-00001', 'desc', 'high', 7.5, '[]', 1, '2023-01-01')" + ) + conn.commit() + finally: + conn.close() + + monkeypatch.setattr( + cve_db_manager.requests, + "get", + lambda url, timeout=60: _FakeResponse(json_data={"vulnerabilities": []}), + ) + + update_kev_data() + + conn = cve_db_manager._connect() + try: + row = conn.execute( + "SELECT kev, kev_date_added FROM cve_entries WHERE cve_id = 'CVE-2024-00001'" + ).fetchone() + finally: + conn.close() + assert row == (0, None) + + +def test_update_epss_data_updates_known_cve_only(monkeypatch, tmp_path: Path) -> None: + db_path = tmp_path / "cve.sqlite" + monkeypatch.setattr(cve_db_manager, "CVE_DB_PATH", str(db_path)) + init_cve_database() + conn = cve_db_manager._connect() + try: + conn.execute( + "INSERT INTO cve_entries (cve_id, description, severity, cvss_score, \"references\") " + "VALUES ('CVE-2024-00001', 'desc', 'high', 7.5, '[]')" + ) + conn.commit() + finally: + conn.close() + + csv_body = ( + "#model_version:v2023.03.01,score_date:2024-01-01T00:00:00+0000\n" + "cve,epss,percentile\n" + "CVE-2024-00001,0.42,0.91\n" + # A CVE we have no local record for is simply not applied. + "CVE-2024-99999,0.10,0.20\n" + ) + monkeypatch.setattr( + cve_db_manager.requests, + "get", + lambda url, timeout=60: _FakeResponse(content=gzip.compress(csv_body.encode("utf-8"))), + ) + + updated = update_epss_data() + assert updated == 1 + + conn = cve_db_manager._connect() + try: + row = conn.execute( + "SELECT epss_score, epss_percentile FROM cve_entries WHERE cve_id = 'CVE-2024-00001'" + ).fetchone() + finally: + conn.close() + assert row == (0.42, 0.91) + + +def test_query_cves_surfaces_kev_and_epss(monkeypatch, tmp_path: Path) -> None: + db_path = _query_cves_db( + tmp_path, + monkeypatch, + [{"version_start": "2.4.2", "version_end": "2.4.10"}], + ) + conn = cve_db_manager._connect() + try: + conn.execute( + "UPDATE cve_entries SET kev = 1, kev_date_added = '2024-02-01', " + "epss_score = 0.9, epss_percentile = 0.99 WHERE cve_id = 'CVE-2024-00001'" + ) + conn.commit() + finally: + conn.close() + + matches = query_cves("apache", version="2.4.9") + assert len(matches) == 1 + assert matches[0]["kev"] is True + assert matches[0]["kev_date_added"] == "2024-02-01" + assert matches[0]["epss_score"] == 0.9 + assert matches[0]["epss_percentile"] == 0.99 + + +def test_ensure_enrichment_columns_migrates_pre_existing_db(monkeypatch, tmp_path: Path) -> None: + # Simulate a database built before the kev/epss columns existed. + db_path = tmp_path / "cve.sqlite" + monkeypatch.setattr(cve_db_manager, "CVE_DB_PATH", str(db_path)) + conn = cve_db_manager._connect() + try: + conn.execute( + """ + CREATE TABLE cve_entries ( + cve_id TEXT PRIMARY KEY, + description TEXT NOT NULL, + severity TEXT, + cvss_score REAL, + cvss_vector TEXT, + published_date TEXT, + last_modified TEXT, + "references" TEXT + ) + """ + ) + conn.commit() + finally: + conn.close() + + init_cve_database() + + conn = cve_db_manager._connect() + try: + columns = {row[1] for row in conn.execute("PRAGMA table_info(cve_entries)")} + finally: + conn.close() + assert {"kev", "kev_date_added", "epss_score", "epss_percentile"} <= columns