feat(evaulator): store package_name and evr for all cve types + evr cache in evaluator

Dockerfile

@@ -59,11 +59,11 @@
 ENV prometheus_multiproc_dir=/tmp/prometheus_multiproc
 
 # minimal schema required by application, used for waiting in services until DB migration is finished
-ENV MINIMAL_SCHEMA=158
+ENV MINIMAL_SCHEMA=172
 
 # Baked-in content for FedRAMP
 ARG STATIC_ASSETS=0
 ARG GIT_TOKEN=""
 RUN if [ "${STATIC_ASSETS}" == 1 ] ; then \
         curl -o /etc/pki/ca-trust/source/anchors/2022-IT-Root-CA.crt https://certs.corp.redhat.com/certs/2022-IT-Root-CA.pem && \
         update-ca-trust extract && \

common/constants.py

@@ -42,6 +42,22 @@ class HostType(StrEnum):
     NONE = "none"
 
 
+class Arch(StrEnum):
+    """Values accepted by the database arch enum."""
+
+    AARCH64 = "aarch64"
+    I386 = "i386"
+    I686 = "i686"
+    NOARCH = "noarch"
+    PPC = "ppc"
+    PPC64 = "ppc64"
+    PPC64LE = "ppc64le"
+    S390 = "s390"
+    S390X = "s390x"
+    SRC = "src"
+    X86_64 = "x86_64"
+
+
 class EvaluatorMessageType(StrEnum):
     """Message types which can arrive at kafka"""
 

database/schema/local_init_db.sh

@@ -12,5 +12,12 @@ until pg_isready -h "${POSTGRES_HOST}" -p "${POSTGRES_PORT}" -U "${POSTGRES_USER
 do sleep 2;
 done
 
-echo "Inserting mock data."
-psql_exec ./database/schema/ve_db_dev_data.sql
+# Try to initialize local schema, if there are no data
+EXISTING_DATA=$(echo "select 1 from system_platform limit 1" | psql_exec - | sed 's/[[:space:]]//g')
+RETVAL=$?
+if [[ "$RETVAL" == "0" && "$EXISTING_DATA" != "1" ]]; then
+    echo "Inserting mock data."
+    psql_exec ./database/schema/ve_db_dev_data.sql
+else
+    echo "Skipping mock data insert, some systems already present."
+fi

evaluator/common.py

@@ -20,6 +20,8 @@
 CFG = Config()
 PROMETHEUS_PORT = CFG.prometheus_port or str(CFG.evaluator_prometheus_port)
 PACKAGE_RE = re.compile(r"(([0-9]+):)?(?P<pn>[^:]+)-.*[^-:]+-.*")
+EVRA_RE = re.compile(r"((?P<epoch>[0-9]+):)?(?P<version>[^-:]+)-(?P<release>[^-:]+)\.(?P<arch>[a-z0-9_]+)")
+
 
 # Prometheus timings
 VMAAS_EVAL_TIME = Histogram(
@@ -61,16 +63,20 @@
 CveCache = namedtuple("CveCache", ["id", "impact_id", "exploitable"])
 # single member inside package name cache
 PackageNameCache = namedtuple("PackageNameCache", ["id"])
+# single member inside EVR cache
+EvrCache = namedtuple("EvrCache", ["id"])
+# parsed EVRA data from VMAAS
+Evra = namedtuple("Evra", ["epoch", "version", "release", "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, from vmaas
-CveAdvisories = namedtuple("Cve", ["name", "advisories"])
-# cve coupled with its package_name, cpe and module, from vmaas
-CveUnpatched = namedtuple("Cve", ["cve", "package_name", "cpe", "module_name", "module_stream"])
+# cve coupled with its advisories and affected package-to-EVRA 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"])
 
 
 # system_platform row taken from DB

evaluator/logic.py

@@ -19,20 +19,24 @@
 from psycopg_pool.pool_async import AsyncConnectionPool
 
 from common.constants import TIMESTAMP_LAST_CVE_SYNC
+from common.constants import Arch
 from common.constants import format_vmaas_cve_endpoint
 from common.logging import get_logger
 from common.peewee_model import VulnerabilityState
 from common.vmaas_client import vmaas_request
 
 from .common import CFG
 from .common import EVAL_PART_TIME
+from .common import EVRA_RE
 from .common import RULES_EVAL_TIME
 from .common import VMAAS_EVAL_TIME
 from .common import CpeCache
 from .common import CveAdvisories
 from .common import CveCache
 from .common import CveImpactCache
 from .common import CveUnpatched
+from .common import Evra
+from .common import EvrCache
 from .common import ModuleCache
 from .common import PackageNameCache
 from .common import RuleCache
@@ -54,6 +58,7 @@ def __init__(self, db_pool: AsyncConnectionPool):
         self.cve_impact_cache: Dict[str, CveImpactCache] = {}
         self.cve_cache: Dict[str, CveCache] = {}
         self.package_name_cache: Dict[str, PackageNameCache] = {}
+        self.evr_cache: Dict[Tuple[int, str, str], EvrCache] = {}
         self.cpe_cache: Dict[str, CpeCache] = {}
         self.module_cache: Dict[str, ModuleCache] = {}
         self.vulnerable_package_cache: Dict[(int, int, Optional[int]), VulnerablePackageCache] = {}
@@ -69,6 +74,7 @@ async def init(self):
         self.cve_impact_cache = await self._load_cve_impact_cache()
         self.cve_cache = await self._load_cve_cache()
         self.package_name_cache = await self._load_package_name_cache()
+        self.evr_cache = await self._load_evr_cache()
         self.cpe_cache = await self._load_cpe_cache()
         self.module_cache = await self._load_module_cache()
         self.vulnerable_package_cache = await self._load_vulnerable_package_cache()
@@ -176,6 +182,16 @@ async def _load_package_name_cache(self) -> Dict[str, PackageNameCache]:
                     cache[package_name["name"]] = PackageNameCache(package_name["id"])
         return cache
 
+    async def _load_evr_cache(self) -> Dict[Tuple[int, str, str], EvrCache]:
+        """Load EVR cache from DB"""
+        cache = {}
+        async with self.db_pool.connection() as conn:
+            async with conn.cursor(row_factory=dict_row) as cur:
+                await cur.execute("""SELECT id, epoch, version, release FROM evr""")
+                for evr in await cur.fetchall():
+                    cache[(evr["epoch"], evr["version"], evr["release"])] = EvrCache(evr["id"])
+        return cache
+
     async def _load_cpe_cache(self) -> Dict[str, CpeCache]:
         """Load cpe cache from DB"""
         cache = {}
@@ -292,6 +308,22 @@ async def _insert_package_name(self, package_name: str):
                     row = await cur.fetchone()
             self.package_name_cache[row["name"]] = PackageNameCache(row["id"])
 
+    async def _insert_evr(self, epoch: int, version: str, release: str):
+        """Insert EVR into database and add it to the cache"""
+        async with self.db_pool.connection() as conn:
+            async with conn.transaction():
+                async with conn.cursor(row_factory=dict_row) as cur:
+                    await cur.execute(
+                        """INSERT INTO evr (epoch, version, release)
+                                        VALUES (%s, %s, %s)
+                                        ON CONFLICT (epoch, version, release) DO UPDATE
+                                        SET epoch = %s, version = %s, release = %s
+                                        RETURNING id, epoch, version, release""",
+                        (epoch, version, release, epoch, version, release),
+                    )
+                    row = await cur.fetchone()
+            self.evr_cache[(row["epoch"], row["version"], row["release"])] = EvrCache(row["id"])
+
     async def _insert_cpe(self, cpe: str):
         """Insert CPE into database and add it to the cache"""
         async with self.db_pool.connection() as conn:
@@ -358,6 +390,33 @@ 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"""
+        parsed_evra = self._parse_evra(evra)
+        if not parsed_evra:
+            return None
+
+        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]
+
+    @staticmethod
+    def _parse_evra(evra: str) -> Optional[Evra]:
+        """Parse EVRA data returned by VMAAS and validate its architecture"""
+        match = EVRA_RE.fullmatch(evra)
+        if not match:
+            LOGGER.warning("unable to parse EVRA from VMAAS: %s", evra)
+            return None
+
+        try:
+            arch = Arch(match.group("arch"))
+        except ValueError:
+            LOGGER.warning("unsupported architecture in VMAAS EVRA: %s", evra)
+            return None
+
+        return Evra(int(match.group("epoch") or 0), match.group("version"), match.group("release"), arch)
+
     async def _get_or_upsert_cpe(self, cpe: str) -> CpeCache:
         """Returns CPE from cache, or inserts the CPE into DB and cache"""
         if cpe not in self.cpe_cache:
@@ -379,7 +438,7 @@ async def _get_or_upsert_vulnerable_package(
         return self.vulnerable_package_cache[(package_name_id, cpe_id, module_id)]
 
     @time(EVAL_PART_TIME.labels(part="vmaas_request"))
-    async def _perform_vmaas_request(self, vmaas_json: dict) -> (List[CveAdvisories], List[CveAdvisories], List[CveUnpatched]):
+    async def _perform_vmaas_request(self, vmaas_json: dict) -> Tuple[List[CveAdvisories], List[CveAdvisories], List[CveUnpatched]]:
         """Perform VMAAS request for package based evaluation"""
         playbook_cves = []
         manually_fixable_cves = []
@@ -394,11 +453,20 @@ async def _perform_vmaas_request(self, vmaas_json: dict) -> (List[CveAdvisories]
         vmaas_response = await vmaas_request(CFG.vmaas_vulnerabilities_endpoint, vmaas_json)
         if vmaas_response:
             playbook_cves = [
-                CveAdvisories(cve["cve"], ",".join(sorted(cve["errata"] or [])) or None) for cve in vmaas_response.get("cve_list", [])
+                CveAdvisories(
+                    cve["cve"],
+                    ",".join(sorted(cve["errata"] or [])) or None,
+                    {affected_package["package_name"]: affected_package["evra"] for affected_package in cve.get("affected", [])},
+                )
+                for cve in vmaas_response.get("cve_list", [])
             ]
 
             manually_fixable_cves = [
-                CveAdvisories(cve["cve"], ",".join(sorted(cve["errata"] or [])) or None)
+                CveAdvisories(
+                    cve["cve"],
+                    ",".join(sorted(cve["errata"] or [])) or None,
+                    {affected_package["package_name"]: affected_package["evra"] for affected_package in cve.get("affected", [])},
+                )
                 for cve in vmaas_response.get("manually_fixable_cve_list", [])
             ]
 
@@ -407,6 +475,7 @@ async def _perform_vmaas_request(self, vmaas_json: dict) -> (List[CveAdvisories]
                 CveUnpatched(
                     cve["cve"],
                     affected_package["package_name"],
+                    affected_package["evra"],
                     affected_package["cpe"],
                     affected_package["module_name"],
                     affected_package["module_stream"],
@@ -563,6 +632,9 @@ 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)
                 sys_vuln_rows[cve_adv.name] = SystemVulnerabilitiesRow(
                     VulnerabilityState.VULNERABLE_BY_PACKAGE,
                     system_platform.rh_account_id,
@@ -580,6 +652,9 @@ async def _evaluate_vmaas_res(
 
             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)
                 sys_vuln_rows[cve_adv.name] = SystemVulnerabilitiesRow(
                     VulnerabilityState.VULNERABLE_BY_PACKAGE,
                     system_platform.rh_account_id,
@@ -599,6 +674,7 @@ async def _evaluate_vmaas_res(
             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)