diff --git a/common/feature_flags.py b/common/feature_flags.py index da7c3c5af..4126c213e 100644 --- a/common/feature_flags.py +++ b/common/feature_flags.py @@ -22,6 +22,7 @@ OS_EXPOSURE_REPORT_FEATURE = "vulnerability.os_exposure_report" NEW_NOTIFICATIONS_FEATURE = "vulnerability.new_notifications" TIMING_LOG_FEATURE = "vulnerability.timing_log" +EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE = "vulnerability.evaluator_system_vulnerabilities_pkg_detail" class UnleashClientProxy: diff --git a/develfeatureflags.json b/develfeatureflags.json index b7c00538b..21ada618e 100644 --- a/develfeatureflags.json +++ b/develfeatureflags.json @@ -56,6 +56,20 @@ ], "strategy": "default", "parameters": {} + }, + { + "name": "vulnerability.evaluator_system_vulnerabilities_pkg_detail", + "type": "release", + "enabled": true, + "stale": false, + "strategies": [ + { + "name": "default", + "parameters": {} + } + ], + "strategy": "default", + "parameters": {} } ] } diff --git a/evaluator/common.py b/evaluator/common.py index 4a606e45e..1007ab12b 100644 --- a/evaluator/common.py +++ b/evaluator/common.py @@ -6,6 +6,7 @@ import re from collections import namedtuple from dataclasses import dataclass +from dataclasses import field from datetime import datetime from typing import Dict from typing import Optional @@ -67,13 +68,20 @@ EvrCache = namedtuple("EvrCache", ["id"]) # parsed EVRA data from VMAAS Evra = namedtuple("Evra", ["epoch", "version", "release", "arch"]) +# affected and fixed EVRAs returned by VMAAS for a package +AffectedPackageEvras = namedtuple("AffectedPackageEvras", ["affected_evra", "fixed_evra"]) +# resolved package data stored for a system vulnerability +SystemVulnerabilityPackageDetail = namedtuple( + "SystemVulnerabilityPackageDetail", + ["package_name_id", "affected_evr_id", "affected_arch", "fixed_evr_id", "fixed_arch"], +) # single member inside cpe cache CpeCache = namedtuple("CpeCache", ["id"]) # single member inside module cache ModuleCache = namedtuple("ModuleCache", ["id"]) # single member inside vulnerable package cache VulnerablePackageCache = namedtuple("VulnerablePackageCache", ["id"]) -# cve coupled with its advisories and affected package-to-EVRA mapping, from vmaas +# cve coupled with its advisories and affected package-to-EVRA-pair mapping, from vmaas CveAdvisories = namedtuple("Cve", ["name", "advisories", "affected_packages"]) # cve coupled with its package name, EVRA, CPE and module, from vmaas CveUnpatched = namedtuple("Cve", ["cve", "package_name", "evra", "cpe", "module_name", "module_stream"]) @@ -139,6 +147,7 @@ class SystemVulnerabilitiesRow: mitigation_reason: str remediation_type_id: int = -1 + affected_package_details: set[SystemVulnerabilityPackageDetail] = field(default_factory=set) def _populate_remediation_type(self): """Populate the remediation type id field""" diff --git a/evaluator/logic.py b/evaluator/logic.py index ac4afa5b7..9b563833e 100644 --- a/evaluator/logic.py +++ b/evaluator/logic.py @@ -21,6 +21,8 @@ from common.constants import TIMESTAMP_LAST_CVE_SYNC from common.constants import Arch from common.constants import format_vmaas_cve_endpoint +from common.feature_flags import EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE +from common.feature_flags import UNLEASH from common.logging import get_logger from common.peewee_model import VulnerabilityState from common.vmaas_client import vmaas_request @@ -30,6 +32,7 @@ from .common import EVRA_RE from .common import RULES_EVAL_TIME from .common import VMAAS_EVAL_TIME +from .common import AffectedPackageEvras from .common import CpeCache from .common import CveAdvisories from .common import CveCache @@ -42,6 +45,7 @@ from .common import RuleCache from .common import SystemPlatform from .common import SystemVulnerabilitiesRow +from .common import SystemVulnerabilityPackageDetail from .common import VmaasErrorException from .common import VulnerablePackageCache @@ -390,8 +394,8 @@ async def _get_or_upsert_package_name(self, package_name: str) -> PackageNameCac await self._insert_package_name(package_name) return self.package_name_cache[package_name] - async def _get_or_upsert_evra(self, evra: str) -> Optional[EvrCache]: - """Parse an EVRA and return its EVR from cache, or insert it into DB and cache""" + async def _get_or_upsert_evra(self, evra: str) -> Optional[Tuple[EvrCache, Arch]]: + """Parse an EVRA and return its cached EVR and architecture""" parsed_evra = self._parse_evra(evra) if not parsed_evra: return None @@ -399,7 +403,31 @@ async def _get_or_upsert_evra(self, evra: str) -> Optional[EvrCache]: key = (parsed_evra.epoch, parsed_evra.version, parsed_evra.release) if key not in self.evr_cache: await self._insert_evr(*key) - return self.evr_cache[key] + return self.evr_cache[key], parsed_evra.arch + + async def _get_affected_package_details( + self, affected_packages: Dict[str, AffectedPackageEvras] + ) -> set[SystemVulnerabilityPackageDetail]: + """Resolve affected package EVRAs to package-detail rows""" + package_details = set() + for package_name, evras in affected_packages.items(): + package_name_cache = await self._get_or_upsert_package_name(package_name) + affected_evr = await self._get_or_upsert_evra(evras.affected_evra) + if not affected_evr: + continue + affected_evr_cache, affected_arch = affected_evr + + fixed_evr_id, fixed_arch = None, None + if evras.fixed_evra: + fixed_evr = await self._get_or_upsert_evra(evras.fixed_evra) + if fixed_evr: + fixed_evr_cache, fixed_arch = fixed_evr + fixed_evr_id = fixed_evr_cache.id + + package_details.add( + SystemVulnerabilityPackageDetail(package_name_cache.id, affected_evr_cache.id, affected_arch, fixed_evr_id, fixed_arch) + ) + return package_details @staticmethod def _parse_evra(evra: str) -> Optional[Evra]: @@ -456,7 +484,10 @@ async def _perform_vmaas_request(self, vmaas_json: dict) -> Tuple[List[CveAdviso CveAdvisories( cve["cve"], ",".join(sorted(cve["errata"] or [])) or None, - {affected_package["package_name"]: affected_package["evra"] for affected_package in cve.get("affected", [])}, + { + affected_package["package_name"]: AffectedPackageEvras(affected_package["evra"], affected_package.get("fixed_evra")) + for affected_package in cve.get("affected", []) + }, ) for cve in vmaas_response.get("cve_list", []) ] @@ -465,7 +496,10 @@ async def _perform_vmaas_request(self, vmaas_json: dict) -> Tuple[List[CveAdviso CveAdvisories( cve["cve"], ",".join(sorted(cve["errata"] or [])) or None, - {affected_package["package_name"]: affected_package["evra"] for affected_package in cve.get("affected", [])}, + { + affected_package["package_name"]: AffectedPackageEvras(affected_package["evra"], None) + for affected_package in cve.get("affected", []) + }, ) for cve in vmaas_response.get("manually_fixable_cve_list", []) ] @@ -553,53 +587,111 @@ async def _delete_vulnerable_package_cve(self, to_delete: List[Tuple[int, int]], ) @time(EVAL_PART_TIME.labels(part="load_system_vulnerable_package")) - async def _load_db_system_vulnerable_package(self, system_platform: SystemPlatform, conn: AsyncConnection) -> Set[int]: - """Make map from system, vulnerable_package pairs in db""" - pkgs = set() + async def _load_db_system_vulnerable_package( + self, system_platform: SystemPlatform, conn: AsyncConnection + ) -> Dict[int, Optional[Tuple[Optional[int], Optional[str]]]]: + """Make map from system vulnerable packages to their affected EVRAs in db""" + pkgs = {} async with conn.cursor(row_factory=dict_row) as cur: - await cur.execute( - """ - SELECT vulnerable_package_id - FROM system_vulnerable_package - WHERE rh_account_id = %s - AND system_id = %s - """, - (system_platform.rh_account_id, system_platform.id), - ) + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE): + await cur.execute( + """ + SELECT vulnerable_package_id, affected_evr_id, affected_arch + FROM system_vulnerable_package + WHERE rh_account_id = %s + AND system_id = %s + """, + (system_platform.rh_account_id, system_platform.id), + ) + else: + await cur.execute( + """ + SELECT vulnerable_package_id + FROM system_vulnerable_package + WHERE rh_account_id = %s + AND system_id = %s + """, + (system_platform.rh_account_id, system_platform.id), + ) for system_vuln_pkg in await cur.fetchall(): - pkgs.add(system_vuln_pkg["vulnerable_package_id"]) + pkgs[system_vuln_pkg["vulnerable_package_id"]] = ( + (system_vuln_pkg["affected_evr_id"], system_vuln_pkg["affected_arch"]) + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE) + else None + ) return pkgs @time(EVAL_PART_TIME.labels(part="cmp_system_vulnerable_package")) async def _compare_system_vulnerable_package( - self, system_platform: SystemPlatform, system_vuln_package_db: Set[int], system_vuln_package: List[int] - ) -> Tuple[List, List]: + self, + system_platform: SystemPlatform, + system_vuln_package_db: Dict[int, Optional[Tuple[Optional[int], Optional[str]]]], + system_vuln_package: Dict[int, Tuple[Optional[int], Optional[str]]], + ) -> Tuple[List, List, List]: """Compare system vulnerable packages in DB with fresh evaluated result""" - to_insert, to_delete = [], [] - for vuln_package_id in system_vuln_package: + to_insert, to_update, to_delete = [], [], [] + for vuln_package_id, affected_evra in system_vuln_package.items(): # vulnerability is not in db, we need to insert it if vuln_package_id not in system_vuln_package_db: - to_insert.append((system_platform.rh_account_id, system_platform.id, vuln_package_id)) + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE): + to_insert.append((system_platform.rh_account_id, system_platform.id, vuln_package_id, *affected_evra)) + else: + to_insert.append((system_platform.rh_account_id, system_platform.id, vuln_package_id)) else: - system_vuln_package_db.remove(vuln_package_id) + affected_evra_db = system_vuln_package_db.pop(vuln_package_id) + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE) and affected_evra != affected_evra_db: + to_update.append((system_platform.rh_account_id, system_platform.id, vuln_package_id, *affected_evra)) # remaining system_vuln_package_db set are vulnerable packages we do not know anything about anymore for vuln_package_id in system_vuln_package_db: to_delete.append((system_platform.rh_account_id, system_platform.id, vuln_package_id)) - return sorted(to_insert), sorted(to_delete) + return sorted(to_insert), sorted(to_update), sorted(to_delete) @time(EVAL_PART_TIME.labels(part="insert_system_vulnerable_package")) - async def _insert_system_vulnerable_package(self, to_insert: List[Tuple[int, int, int]], conn: AsyncConnection): + async def _insert_system_vulnerable_package(self, to_insert: List[Tuple], conn: AsyncConnection): """Insert given system vulnerable packages to table""" + async with conn.cursor() as cur: + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE): + await cur.executemany( + """INSERT INTO system_vulnerable_package + (rh_account_id, system_id, vulnerable_package_id, affected_evr_id, affected_arch, first_reported) + VALUES (%s, %s, %s, %s, %s, now()) + ON CONFLICT DO NOTHING + """, + to_insert, + ) + else: + await cur.executemany( + """INSERT INTO system_vulnerable_package + (rh_account_id, system_id, vulnerable_package_id, first_reported) + VALUES (%s, %s, %s, now()) + ON CONFLICT DO NOTHING + """, + to_insert, + ) + + @time(EVAL_PART_TIME.labels(part="update_system_vulnerable_package")) + async def _update_system_vulnerable_package( + self, to_update: List[Tuple[int, int, int, Optional[int], Optional[str]]], conn: AsyncConnection + ): + """Update affected EVRAs for existing system vulnerable packages""" + if not UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE): + return + async with conn.cursor() as cur: await cur.executemany( - """INSERT INTO system_vulnerable_package - (rh_account_id, system_id, vulnerable_package_id, first_reported) - VALUES (%s, %s, %s, now()) - ON CONFLICT DO NOTHING - """, - to_insert, + """UPDATE system_vulnerable_package + SET affected_evr_id = %s, + affected_arch = %s + WHERE rh_account_id = %s + AND system_id = %s + AND vulnerable_package_id = %s + """, + [ + (affected_evr_id, affected_arch, rh_account_id, system_id, vulnerable_package_id) + for rh_account_id, system_id, vulnerable_package_id, affected_evr_id, affected_arch in to_update + ], ) @time(EVAL_PART_TIME.labels(part="delete_system_vulnerable_package")) @@ -632,9 +724,7 @@ async def _evaluate_vmaas_res( # system is potentially vulnerable to cves returned from vmaas for cve_adv in playbook_cves: cve = await self._get_or_upsert_cve(cve_adv.name) - for package_name, evra in cve_adv.affected_packages.items(): - await self._get_or_upsert_package_name(package_name) - await self._get_or_upsert_evra(evra) + affected_package_details = await self._get_affected_package_details(cve_adv.affected_packages) sys_vuln_rows[cve_adv.name] = SystemVulnerabilitiesRow( VulnerabilityState.VULNERABLE_BY_PACKAGE, system_platform.rh_account_id, @@ -648,13 +738,12 @@ async def _evaluate_vmaas_res( None, None, None, + affected_package_details=affected_package_details, ) for cve_adv in manually_fixable_cves: cve = await self._get_or_upsert_cve(cve_adv.name) - for package_name, evra in cve_adv.affected_packages.items(): - await self._get_or_upsert_package_name(package_name) - await self._get_or_upsert_evra(evra) + affected_package_details = await self._get_affected_package_details(cve_adv.affected_packages) sys_vuln_rows[cve_adv.name] = SystemVulnerabilitiesRow( VulnerabilityState.VULNERABLE_BY_PACKAGE, system_platform.rh_account_id, @@ -668,28 +757,38 @@ async def _evaluate_vmaas_res( None, None, None, + affected_package_details=affected_package_details, ) # aggregate unfixed cves, get (package name, cpe, module) tuples and list of cves for these pairs pn_cpes = {} for cve_unpatched in unpatched_cves: cve_cache = await self._get_or_upsert_cve(cve_unpatched.cve) - await self._get_or_upsert_evra(cve_unpatched.evra) - pn_cpes.setdefault( - (cve_unpatched.package_name, cve_unpatched.cpe, cve_unpatched.module_name, cve_unpatched.module_stream), set() - ).add(cve_cache.id) - - vuln_package_cve = {} - for (package_name, cpe, module_name, module_stream), cve_ids in pn_cpes.items(): - package_name_cache = await self._get_or_upsert_package_name(package_name) - cpe_cache = await self._get_or_upsert_cpe(cpe) - if module_name and module_stream: - module_cache = await self._get_or_upsert_module(module_name, module_stream) - module_id = module_cache.id - else: - module_id = None - vulnerable_package_cache = await self._get_or_upsert_vulnerable_package(package_name_cache.id, cpe_cache.id, module_id) - vuln_package_cve[vulnerable_package_cache.id] = cve_ids + evr_cache = await self._get_or_upsert_evra(cve_unpatched.evra) + affected_evra = (evr_cache[0].id, evr_cache[1].value) if evr_cache else (None, None) + pn_cpe = ( + cve_unpatched.package_name, + cve_unpatched.cpe, + cve_unpatched.module_name, + cve_unpatched.module_stream, + ) + cve_ids, _ = pn_cpes.setdefault(pn_cpe, (set(), affected_evra)) + cve_ids.add(cve_cache.id) + pn_cpes[pn_cpe] = (cve_ids, affected_evra) + + vuln_package_cve = {} + vuln_package_evras = {} + for (package_name, cpe, module_name, module_stream), (cve_ids, affected_evra) in pn_cpes.items(): + package_name_cache = await self._get_or_upsert_package_name(package_name) + cpe_cache = await self._get_or_upsert_cpe(cpe) + if module_name and module_stream: + module_cache = await self._get_or_upsert_module(module_name, module_stream) + module_id = module_cache.id + else: + module_id = None + vulnerable_package_cache = await self._get_or_upsert_vulnerable_package(package_name_cache.id, cpe_cache.id, module_id) + vuln_package_cve[vulnerable_package_cache.id] = cve_ids + vuln_package_evras[vulnerable_package_cache.id] = affected_evra # ensure all vulnerable_package_cve pairs are up to date vuln_package_cve_db = await self._load_db_vulnerable_package_cve(list(vuln_package_cve), conn) @@ -705,23 +804,29 @@ async def _evaluate_vmaas_res( await self._delete_vulnerable_package_cve(to_delete, conn) # limit number of vulnerable packages linked to a system (skip vulnerable packages affecting same CVEs) - target_system_vuln_package = [] + target_system_vuln_package = {} seen_cves = set() for vuln_package_id, cve_ids in sorted(vuln_package_cve.items()): if not cve_ids.issubset(seen_cves): - target_system_vuln_package.append(vuln_package_id) + target_system_vuln_package[vuln_package_id] = vuln_package_evras[vuln_package_id] seen_cves.update(cve_ids) # ensure all system_vulnerable_package pairs are up to date system_vuln_package_db = await self._load_db_system_vulnerable_package(system_platform, conn) - to_insert, to_delete = await self._compare_system_vulnerable_package( + to_insert, to_update, to_delete = await self._compare_system_vulnerable_package( system_platform, system_vuln_package_db, target_system_vuln_package ) LOGGER.debug( - "system: %s, system_vulnerable_package changes, i: %s, d: %s", system_platform.inventory_id, len(to_insert), len(to_delete) + "system: %s, system_vulnerable_package changes, i: %s, u: %s, d: %s", + system_platform.inventory_id, + len(to_insert), + len(to_update), + len(to_delete), ) if to_insert: await self._insert_system_vulnerable_package(to_insert, conn) + if to_update: + await self._update_system_vulnerable_package(to_update, conn) if to_delete: await self._delete_system_vulnerable_package(to_delete, conn) diff --git a/evaluator/processor.py b/evaluator/processor.py index 8a96a251a..0a497bc0a 100644 --- a/evaluator/processor.py +++ b/evaluator/processor.py @@ -19,6 +19,8 @@ from psycopg.rows import dict_row from psycopg_pool.pool_async import AsyncConnectionPool +from common.feature_flags import EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE +from common.feature_flags import UNLEASH from common.logging import get_logger from common.mqueue import MQWriter from common.peewee_model import VulnerabilityState @@ -270,6 +272,96 @@ async def _delete_vulnerabilities(self, to_delete: [int], system_platform: Syste (to_delete, system_platform.rh_account_id), ) + @time(EVAL_PART_TIME.labels(part="load_system_vulnerability_package_details")) + async def _load_db_system_vulnerability_package_details( + self, system_platform: SystemPlatform, system_vulnerability_ids: List[int], conn: AsyncConnection + ) -> set[Tuple[int, int, int, str, Optional[int], Optional[str]]]: + """Load existing package details for active vulnerabilities of a system""" + if not system_vulnerability_ids: + return set() + + package_details = set() + async with conn.cursor(row_factory=dict_row) as cur: + await cur.execute( + """ + SELECT svpd.system_vulnerability_id, svpd.package_name_id, svpd.affected_evr_id, svpd.affected_arch, + svpd.fixed_evr_id, svpd.fixed_arch + FROM system_vulnerabilities_pkg_detail AS svpd + WHERE svpd.rh_account_id = %s + AND svpd.system_vulnerability_id = ANY(%s) + """, + (system_platform.rh_account_id, system_vulnerability_ids), + ) + for package_detail in await cur.fetchall(): + package_details.add( + ( + package_detail["system_vulnerability_id"], + package_detail["package_name_id"], + package_detail["affected_evr_id"], + package_detail["affected_arch"], + package_detail["fixed_evr_id"], + package_detail["fixed_arch"], + ) + ) + return package_details + + @time(EVAL_PART_TIME.labels(part="sync_system_vulnerability_package_details")) + async def _sync_system_vulnerability_package_details( + self, + system_platform: SystemPlatform, + vulnerabilities: Dict[str, SystemVulnerabilitiesRow], + system_vulnerability_ids: Dict[int, int], + conn: AsyncConnection, + ): + """Insert missing and remove redundant affected package details for a system""" + package_details_db = await self._load_db_system_vulnerability_package_details( + system_platform, list(system_vulnerability_ids.values()), conn + ) + package_details = set() + for vulnerability in vulnerabilities.values(): + vulnerability_id = system_vulnerability_ids.get(vulnerability.cve_id) + if vulnerability_id is None: + continue + for package_detail in vulnerability.affected_package_details: + package_details.add( + ( + vulnerability_id, + package_detail.package_name_id, + package_detail.affected_evr_id, + package_detail.affected_arch.value, + package_detail.fixed_evr_id, + package_detail.fixed_arch.value if package_detail.fixed_arch else None, + ) + ) + + to_delete = package_details_db - package_details + to_insert = package_details - package_details_db + LOGGER.debug( + "system: %s, system vulnerability package detail changes, i: %s, d: %s", + system_platform.inventory_id, + len(to_insert), + len(to_delete), + ) + async with conn.cursor() as cur: + if to_delete: + await cur.executemany( + """DELETE FROM system_vulnerabilities_pkg_detail + WHERE rh_account_id = %s + AND system_vulnerability_id = %s + AND package_name_id = %s + """, + [(system_platform.rh_account_id, package_detail[0], package_detail[1]) for package_detail in to_delete], + ) + if to_insert: + await cur.executemany( + """INSERT INTO system_vulnerabilities_pkg_detail + (rh_account_id, system_vulnerability_id, package_name_id, affected_evr_id, affected_arch, + fixed_evr_id, fixed_arch) + VALUES (%s, %s, %s, %s, %s, %s, %s) + """, + [(system_platform.rh_account_id, *package_detail) for package_detail in to_insert], + ) + @time(EVAL_PART_TIME.labels(part="mark_system_evaluated")) async def _mark_system_evaluated(self, cve_cache_count: int, system_platform: SystemPlatform, conn: AsyncConnection): """Mark system as evaluated, updating its cache and last evaluation""" @@ -344,6 +436,11 @@ async def _evaluate_system( self.evaluator_logic.evaluate_vulnerabilities(system_platform, conn), ) + system_vulnerability_ids = { + sys_vuln_rows[cve].cve_id: vulnerability_db.sv_id + for cve, vulnerability_db in sys_vuln_rows_db.items() + if cve in sys_vuln_rows + } to_insert, to_update, to_delete = await self._compare_sys_vulns(sys_vuln_rows_db, sys_vuln_rows) LOGGER.debug( "system: %s, system_vulnerabilities changes, i: %s, u: %s, d: %s", @@ -355,10 +452,15 @@ async def _evaluate_system( new_system_vulns = [] if to_insert: new_system_vulns = await self._insert_vulnerabilities(to_insert, conn) + system_vulnerability_ids.update( + {cve_id: system_vulnerability_id for system_vulnerability_id, cve_id in new_system_vulns} + ) if to_update: await self._update_vulnerabilities(to_update, conn) if to_delete: await self._delete_vulnerabilities(to_delete, system_platform, conn) + if UNLEASH.is_enabled(EVALUATOR_SYSTEM_VULNERABILITIES_PKG_DETAIL_FEATURE): + await self._sync_system_vulnerability_package_details(system_platform, sys_vuln_rows, system_vulnerability_ids, conn) fixable_sys_vuln_rows = {} total_cves = 0