From c2e6f00e66559c2c1cfac6343712f81cd8070d27 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Mon, 19 Feb 2024 18:54:25 +0100 Subject: [PATCH 01/14] fix(csaf): finalize db schema RHINENG-7862 --- database/upgrade_scripts/020-csaf_tables2.sql | 68 +++++++++++++++++++ database/vmaas_db_postgresql.sql | 62 +++++++++++------ 2 files changed, 111 insertions(+), 19 deletions(-) create mode 100644 database/upgrade_scripts/020-csaf_tables2.sql diff --git a/database/upgrade_scripts/020-csaf_tables2.sql b/database/upgrade_scripts/020-csaf_tables2.sql new file mode 100644 index 000000000..7aa5f64ec --- /dev/null +++ b/database/upgrade_scripts/020-csaf_tables2.sql @@ -0,0 +1,68 @@ +DROP TABLE csaf_cves; +DROP TABLE csaf_products; +UPDATE csaf_file SET updated = NULL; + +-- ----------------------------------------------------- +-- Table vmaas.csaf_product +-- ----------------------------------------------------- +CREATE TABLE IF NOT EXISTS csaf_product ( + id SERIAL, + cpe_id INT NOT NULL, + package_name_id INT NULL, + package_id INT NULL, + module_stream TEXT NULL CHECK (NOT empty(module_stream)), + PRIMARY KEY (id), + CONSTRAINT cpe_id + FOREIGN KEY (cpe_id) + REFERENCES cpe (id), + CONSTRAINT package_name_id + FOREIGN KEY (package_name_id) + REFERENCES package_name (id), + CONSTRAINT package_id + FOREIGN KEY (package_id) + REFERENCES package (id), + CONSTRAINT pkg_id CHECK( + (package_name_id IS NOT NULL AND package_id IS NULL) OR + (package_name_id IS NULL AND package_id IS NOT NULL)) +)TABLESPACE pg_default; + +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_name_id) WHERE package_name_id IS NOT NULL AND package_id IS NULL AND module_stream IS NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_id) WHERE package_id IS NOT NULL AND package_name_id IS NULL AND module_stream IS NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_name_id, module_stream) WHERE package_name_id IS NOT NULL AND package_id IS NULL AND module_stream IS NOT NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_id, module_stream) WHERE package_id IS NOT NULL AND package_name_id IS NULL AND module_stream IS NOT NULL; + +-- ----------------------------------------------------- +-- Table vmaas.csaf_cve_product +-- ----------------------------------------------------- +CREATE TABLE IF NOT EXISTS csaf_cve_product ( + id SERIAL, + cve_id INT NOT NULL, + csaf_product_id INT NOT NULL, + csaf_product_status_id INT NOT NULL, + csaf_file_id INT NOT NULL, + PRIMARY KEY (id), + CONSTRAINT cve_id + FOREIGN KEY (cve_id) + REFERENCES cve (id), + CONSTRAINT csaf_product_id + FOREIGN KEY (csaf_product_id) + REFERENCES csaf_product (id), + CONSTRAINT csaf_product_status_id + FOREIGN KEY (csaf_product_status_id) + REFERENCES csaf_product_status (id), + CONSTRAINT csaf_file_id + FOREIGN KEY (csaf_file_id) + REFERENCES csaf_file (id), + CONSTRAINT csaf_cve_product_ids_uq + UNIQUE (cve_id, csaf_product_id) +)TABLESPACE pg_default; + + +-- ----------------------------------------------------- +-- vmaas users permission setup: +-- vmaas_writer - has rights to INSERT/UPDATE/DELETE; used by reposcan +-- vmaas_reader - has SELECT only; used by webapp +-- ----------------------------------------------------- +GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO vmaas_writer; +GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA public TO vmaas_writer; +GRANT SELECT ON ALL TABLES IN SCHEMA public TO vmaas_reader; diff --git a/database/vmaas_db_postgresql.sql b/database/vmaas_db_postgresql.sql index 7586b5562..efd72f769 100644 --- a/database/vmaas_db_postgresql.sql +++ b/database/vmaas_db_postgresql.sql @@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS db_version ( )TABLESPACE pg_default; -- Increment this when editing this file -INSERT INTO db_version (name, version) VALUES ('schema_version', 19); +INSERT INTO db_version (name, version) VALUES ('schema_version', 20); -- ----------------------------------------------------- -- evr type @@ -1307,34 +1307,58 @@ ON CONFLICT DO NOTHING; -- ----------------------------------------------------- --- Table vmaas.csaf_products +-- Table vmaas.csaf_product -- ----------------------------------------------------- -CREATE TABLE IF NOT EXISTS csaf_products ( - id SERIAL, - cpe TEXT NOT NULL, CHECK (NOT empty(cpe)), - package TEXT NULL, CHECK (NOT empty(package)), - module TEXT NULL, CHECK (NOT empty(module)), - CONSTRAINT unique_csaf_product UNIQUE (cpe, package, module), - PRIMARY KEY (id) +CREATE TABLE IF NOT EXISTS csaf_product ( + id SERIAL, + cpe_id INT NOT NULL, + package_name_id INT NULL, + package_id INT NULL, + module_stream TEXT NULL CHECK (NOT empty(module_stream)), + PRIMARY KEY (id), + CONSTRAINT cpe_id + FOREIGN KEY (cpe_id) + REFERENCES cpe (id), + CONSTRAINT package_name_id + FOREIGN KEY (package_name_id) + REFERENCES package_name (id), + CONSTRAINT package_id + FOREIGN KEY (package_id) + REFERENCES package (id), + CONSTRAINT pkg_id CHECK( + (package_name_id IS NOT NULL AND package_id IS NULL) OR + (package_name_id IS NULL AND package_id IS NOT NULL)) )TABLESPACE pg_default; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_name_id) WHERE package_name_id IS NOT NULL AND package_id IS NULL AND module_stream IS NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_id) WHERE package_id IS NOT NULL AND package_name_id IS NULL AND module_stream IS NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_name_id, module_stream) WHERE package_name_id IS NOT NULL AND package_id IS NULL AND module_stream IS NOT NULL; +CREATE UNIQUE INDEX ON csaf_product(cpe_id, package_id, module_stream) WHERE package_id IS NOT NULL AND package_name_id IS NULL AND module_stream IS NOT NULL; -- ----------------------------------------------------- --- Table vmaas.csaf_cves +-- Table vmaas.csaf_cve_product -- ----------------------------------------------------- -CREATE TABLE IF NOT EXISTS csaf_cves ( - id SERIAL, - cve TEXT NOT NULL, CHECK (NOT empty(cve)), - csaf_product_id INT NOT NULL, - product_status_id INT NOT NULL, - CONSTRAINT unique_csaf_cve_product UNIQUE (cve, csaf_product_id), +CREATE TABLE IF NOT EXISTS csaf_cve_product ( + id SERIAL, + cve_id INT NOT NULL, + csaf_product_id INT NOT NULL, + csaf_product_status_id INT NOT NULL, + csaf_file_id INT NOT NULL, PRIMARY KEY (id), + CONSTRAINT cve_id + FOREIGN KEY (cve_id) + REFERENCES cve (id), CONSTRAINT csaf_product_id FOREIGN KEY (csaf_product_id) - REFERENCES csaf_products (id), + REFERENCES csaf_product (id), CONSTRAINT csaf_product_status_id - FOREIGN KEY (product_status_id) - REFERENCES csaf_product_status (id) + FOREIGN KEY (csaf_product_status_id) + REFERENCES csaf_product_status (id), + CONSTRAINT csaf_file_id + FOREIGN KEY (csaf_file_id) + REFERENCES csaf_file (id), + CONSTRAINT csaf_cve_product_ids_uq + UNIQUE (cve_id, csaf_product_id) )TABLESPACE pg_default; From 3242f221c9b283740a4a511522b098ad2c37564e Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Wed, 6 Mar 2024 16:02:21 +0100 Subject: [PATCH 02/14] fix(csaf): handle exceptions when saving files --- vmaas/reposcan/database/csaf_store.py | 32 ++++++++++++++++----------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/vmaas/reposcan/database/csaf_store.py b/vmaas/reposcan/database/csaf_store.py index cad46a7c6..a8547c78a 100644 --- a/vmaas/reposcan/database/csaf_store.py +++ b/vmaas/reposcan/database/csaf_store.py @@ -32,20 +32,26 @@ def delete_csaf_file(self, name: str): cur.close() def _save_csaf_files(self, csaf_files: CsafFiles) -> list[int]: + res = [] cur = self.conn.cursor() - db_ids = execute_values( - cur, - """ - insert into csaf_file (name, updated) values %s - on conflict (name) do update set updated = excluded.updated - returning id - """, - csaf_files.to_tuples(("name", "csv_timestamp")), - fetch=True, - ) - cur.close() - self.conn.commit() - return db_ids + try: + res = execute_values( + cur, + """ + insert into csaf_file (name, updated) values %s + on conflict (name) do update set updated = excluded.updated + returning id + """, + csaf_files.to_tuples(("name", "csv_timestamp")), + fetch=True, + ) + self.conn.commit() + except Exception as exc: + CSAF_FAILED_IMPORT.inc() + self.logger.exception("Failed to import csaf file to DB: %s", exc) + finally: + cur.close() + return [r[0] for r in res] def _get_product_rows(self, products_by_status) -> tuple[dict[str, list[tuple[str, str, str]]], list[tuple[int, str]]]: to_associate_ids = [] From 18a67a4ad3fc03da4fc87f29a80848c225acb7b8 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Wed, 6 Mar 2024 16:21:43 +0100 Subject: [PATCH 03/14] chore(csaf): move csaf test data to conftest --- vmaas/reposcan/conftest.py | 31 ++++++++++++++ .../redhatcsaf/test/test_controller.py | 42 +++---------------- 2 files changed, 37 insertions(+), 36 deletions(-) diff --git a/vmaas/reposcan/conftest.py b/vmaas/reposcan/conftest.py index 55d023a74..4fa8cf6ce 100644 --- a/vmaas/reposcan/conftest.py +++ b/vmaas/reposcan/conftest.py @@ -8,11 +8,42 @@ from vmaas.common.paths import USER_CREATE_SQL_PATH, DB_CREATE_SQL_PATH from vmaas.reposcan.database.database_handler import init_db +from vmaas.reposcan.redhatcsaf import modeling as csaf_model VMAAS_DIR = Path(__file__).resolve().parent.parent.parent VMAAS_DB_DATA = VMAAS_DIR.joinpath("vmaas", "reposcan", "test_data", "database", "test_data.sql") VMAAS_PG_OLD = VMAAS_DIR.joinpath("vmaas", "reposcan", "test_data", "database", "vmaas_db_postgresql_old.sql") +EXPECTED_CSAF = ( + ("cve-2023-0030.json", csaf_model.CsafCves({"CVE-2023-0030": []})), + ( + "cve-2023-0049.json", + csaf_model.CsafCves( + { + "CVE-2023-0049": [ + csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:6", "vim", 4), + csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:7", "vim", 4), + csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:8", "vim", 4), + csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:9", "vim", 4), + ] + } + ), + ), + ( + "cve-2023-1017.json", + csaf_model.CsafCves( + { + "CVE-2023-1017": [ + csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:8", "libtpms", 4, "virt:rhel"), + csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.2"), + csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.3"), + csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:av"), + ] + } + ), + ), +) + @pytest.fixture(scope="session") def db_conn(): diff --git a/vmaas/reposcan/redhatcsaf/test/test_controller.py b/vmaas/reposcan/redhatcsaf/test/test_controller.py index e509c9e43..34ccc6b9b 100644 --- a/vmaas/reposcan/redhatcsaf/test/test_controller.py +++ b/vmaas/reposcan/redhatcsaf/test/test_controller.py @@ -1,44 +1,14 @@ """Unit tests of csaf_controller.py.""" import pathlib from datetime import datetime -import pytest +import pytest +from vmaas.reposcan.conftest import EXPECTED_CSAF from vmaas.reposcan.redhatcsaf.csaf_controller import CsafController -from vmaas.reposcan.redhatcsaf.modeling import CsafCves, CsafData, CsafFiles +from vmaas.reposcan.redhatcsaf.modeling import CsafData from vmaas.reposcan.redhatcsaf.modeling import CsafFile -from vmaas.reposcan.redhatcsaf.modeling import CsafProduct - - -EXPECTED_PARSE = ( - ("cve-2023-0030.json", CsafCves({"CVE-2023-0030": []})), - ( - "cve-2023-0049.json", - CsafCves( - { - "CVE-2023-0049": [ - CsafProduct("cpe:/o:redhat:enterprise_linux:6", "vim", 4), - CsafProduct("cpe:/o:redhat:enterprise_linux:7", "vim", 4), - CsafProduct("cpe:/o:redhat:enterprise_linux:8", "vim", 4), - CsafProduct("cpe:/o:redhat:enterprise_linux:9", "vim", 4), - ] - } - ), - ), - ( - "cve-2023-1017.json", - CsafCves( - { - "CVE-2023-1017": [ - CsafProduct("cpe:/o:redhat:enterprise_linux:8", "libtpms", 4, "virt:rhel"), - CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.2"), - CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.3"), - CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:av"), - ] - } - ), - ), -) +from vmaas.reposcan.redhatcsaf.modeling import CsafFiles class TestCsafController: @@ -51,7 +21,7 @@ def csaf(self, db_conn): # pylint: disable=unused-argument csaf.tmp_directory = pathlib.Path(__file__).parent.resolve() return csaf - @pytest.mark.parametrize("data", EXPECTED_PARSE, ids=[x[0] for x in EXPECTED_PARSE]) + @pytest.mark.parametrize("data", EXPECTED_CSAF, ids=[x[0] for x in EXPECTED_CSAF]) def test_parse_csaf_file(self, data, csaf): """Test CSAF JSON file parsing.""" csaf_json, expected = data @@ -59,7 +29,7 @@ def test_parse_csaf_file(self, data, csaf): parsed = csaf.parse_csaf_file(csaf_file) assert parsed == expected - @pytest.mark.parametrize("data", EXPECTED_PARSE, ids=[x[0] for x in EXPECTED_PARSE]) + @pytest.mark.parametrize("data", EXPECTED_CSAF, ids=[x[0] for x in EXPECTED_CSAF]) def test_csaf_store(self, data, csaf: CsafController): """Parse CSAF JSON and store its data to DB""" csaf_json, _ = data From 9a5351f94fd257f4d241204cb98e69ea7147fa8d Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Fri, 8 Mar 2024 18:38:43 +0100 Subject: [PATCH 04/14] chore(csaf): make sure CVEs are always uppercase --- vmaas/reposcan/redhatcsaf/csaf_controller.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/vmaas/reposcan/redhatcsaf/csaf_controller.py b/vmaas/reposcan/redhatcsaf/csaf_controller.py index 0ae4fe9cc..d5c1182e6 100644 --- a/vmaas/reposcan/redhatcsaf/csaf_controller.py +++ b/vmaas/reposcan/redhatcsaf/csaf_controller.py @@ -137,7 +137,8 @@ def _parse_vulnerabilities(self, csaf: dict, product_cpe: dict) -> CsafCves: # `vulnerability` can be identified by `cve` or `ids`, we are interested only in those with `cve` continue - unfixed_cves[vulnerability["cve"]] = [] + cve = vulnerability["cve"].upper() + unfixed_cves[cve] = [] for product_status in self.cfg.csaf_product_status_list: status_id = CsafProductStatus[product_status.upper()].value for unfixed in vulnerability["product_status"].get(product_status.lower(), []): @@ -149,7 +150,7 @@ def _parse_vulnerabilities(self, csaf: dict, product_cpe: dict) -> CsafCves: module, pkg_name = rest.split("/", 1) csaf_product = CsafProduct(product_cpe[branch_product], pkg_name, status_id, module) - unfixed_cves[vulnerability["cve"]].append(csaf_product) + unfixed_cves[cve].append(csaf_product) return unfixed_cves From f03a3838c2c788a2b14854ad7debf5dfb031c180 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Fri, 8 Mar 2024 18:56:36 +0100 Subject: [PATCH 05/14] fix(csaf): extend model with CsafProducts --- vmaas/reposcan/conftest.py | 10 +- vmaas/reposcan/redhatcsaf/csaf_controller.py | 3 +- vmaas/reposcan/redhatcsaf/modeling.py | 171 ++++++++++++++++--- 3 files changed, 152 insertions(+), 32 deletions(-) diff --git a/vmaas/reposcan/conftest.py b/vmaas/reposcan/conftest.py index 4fa8cf6ce..7d383b545 100644 --- a/vmaas/reposcan/conftest.py +++ b/vmaas/reposcan/conftest.py @@ -15,17 +15,17 @@ VMAAS_PG_OLD = VMAAS_DIR.joinpath("vmaas", "reposcan", "test_data", "database", "vmaas_db_postgresql_old.sql") EXPECTED_CSAF = ( - ("cve-2023-0030.json", csaf_model.CsafCves({"CVE-2023-0030": []})), + ("cve-2023-0030.json", csaf_model.CsafCves({"CVE-2023-0030": csaf_model.CsafProducts()})), ( "cve-2023-0049.json", csaf_model.CsafCves( { - "CVE-2023-0049": [ + "CVE-2023-0049": csaf_model.CsafProducts([ csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:6", "vim", 4), csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:7", "vim", 4), csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:8", "vim", 4), csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:9", "vim", 4), - ] + ]) } ), ), @@ -33,12 +33,12 @@ "cve-2023-1017.json", csaf_model.CsafCves( { - "CVE-2023-1017": [ + "CVE-2023-1017": csaf_model.CsafProducts([ csaf_model.CsafProduct("cpe:/o:redhat:enterprise_linux:8", "libtpms", 4, "virt:rhel"), csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.2"), csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:8.3"), csaf_model.CsafProduct("cpe:/a:redhat:advanced_virtualization:8::el8", "libtpms", 4, "virt:av"), - ] + ]) } ), ), diff --git a/vmaas/reposcan/redhatcsaf/csaf_controller.py b/vmaas/reposcan/redhatcsaf/csaf_controller.py index d5c1182e6..a889ccf0f 100644 --- a/vmaas/reposcan/redhatcsaf/csaf_controller.py +++ b/vmaas/reposcan/redhatcsaf/csaf_controller.py @@ -21,6 +21,7 @@ from vmaas.reposcan.redhatcsaf.modeling import CsafFile from vmaas.reposcan.redhatcsaf.modeling import CsafFiles from vmaas.reposcan.redhatcsaf.modeling import CsafProduct +from vmaas.reposcan.redhatcsaf.modeling import CsafProducts from vmaas.reposcan.redhatcsaf.modeling import CsafProductStatus CSAF_VEX_BASE_URL = os.getenv("CSAF_VEX_BASE_URL", "https://access.redhat.com/security/data/csaf/beta/vex/") @@ -138,7 +139,7 @@ def _parse_vulnerabilities(self, csaf: dict, product_cpe: dict) -> CsafCves: continue cve = vulnerability["cve"].upper() - unfixed_cves[cve] = [] + unfixed_cves[cve] = CsafProducts() for product_status in self.cfg.csaf_product_status_list: status_id = CsafProductStatus[product_status.upper()].value for unfixed in vulnerability["product_status"].get(product_status.lower(), []): diff --git a/vmaas/reposcan/redhatcsaf/modeling.py b/vmaas/reposcan/redhatcsaf/modeling.py index 3e8a3cf49..98b9cc60a 100644 --- a/vmaas/reposcan/redhatcsaf/modeling.py +++ b/vmaas/reposcan/redhatcsaf/modeling.py @@ -4,13 +4,12 @@ from __future__ import annotations import csv +import typing as t from dataclasses import dataclass from dataclasses import field from datetime import datetime from enum import IntEnum -from typing import ItemsView, Iterator, KeysView -from typing import Optional -from typing import overload +from pathlib import Path import attr @@ -19,6 +18,7 @@ class CsafProductStatus(IntEnum): """CSAF product status enum.""" + FIRST_AFFECTED = 1 FIRST_FIXED = 2 FIXED = 3 @@ -35,7 +35,7 @@ class CsafFile: name: str csv_timestamp: datetime - db_timestamp: Optional[datetime] = None + db_timestamp: datetime | None = None id_: int = 0 @property @@ -53,7 +53,7 @@ class CsafFiles: _files: dict[str, CsafFile] = attr.Factory(dict) @classmethod - def from_table_map_and_csv(cls, table_map: dict[str, tuple[int, datetime]], csv_path: str) -> CsafFiles: + def from_table_map_and_csv(cls, table_map: dict[str, tuple[int, datetime]], csv_path: Path) -> CsafFiles: """Initialize class from CsafStore.csaf_file_map table_map and changes.csv.""" obj = cls() obj._update_from_table_map(table_map) @@ -69,7 +69,7 @@ def _update_from_table_map(self, table_map: dict[str, tuple[int, datetime]]) -> obj.db_timestamp = timestamp self[key] = obj - def _update_from_csv(self, path: str) -> None: + def _update_from_csv(self, path: Path) -> None: """Update files from CSAF changes.csv.""" with open(path, newline="", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, ("name", "timestamp")) @@ -84,7 +84,7 @@ def out_of_date(self) -> filter[CsafFile]: """Filter generator of out of date files.""" return filter(lambda x: x.out_of_date, self) - def to_tuples(self, attributes: tuple[str, ...]) -> list[tuple]: + def to_tuples(self, attributes: tuple[str, ...]) -> list[tuple[int | str | datetime | None, ...]]: """Transform data to list of tuples with chosen attributes.""" res = [] for item in self: @@ -94,15 +94,15 @@ def to_tuples(self, attributes: tuple[str, ...]) -> list[tuple]: res.append(tuple(items)) return res - @overload - def get(self, key: str, default: None = None) -> Optional[CsafFile]: + @t.overload + def get(self, key: str, default: None = None) -> CsafFile | None: ... - @overload + @t.overload def get(self, key: str, default: CsafFile) -> CsafFile: ... - def get(self, key, default=None): + def get(self, key: str, default: CsafFile | None = None) -> CsafFile | None: """Return the value for key if key is in the collection, else default.""" return self._files.get(key, default) @@ -119,7 +119,7 @@ def __setitem__(self, key: str, val: CsafFile) -> None: def __contains__(self, key: str) -> bool: return key in self._files - def __iter__(self) -> Iterator[CsafFile]: + def __iter__(self) -> t.Iterator[CsafFile]: return iter(self._files.values()) def __next__(self) -> CsafFile: @@ -131,6 +131,9 @@ def __repr__(self) -> str: def __len__(self) -> int: return len(self._files) + def __bool__(self) -> bool: + return bool(self._files) + @dataclass class CsafProduct: @@ -139,16 +142,129 @@ class CsafProduct: cpe: str package: str status_id: int - module: Optional[str] = None + module: str | None = None + id_: int | None = None + cpe_id: int | None = None + package_name_id: int | None = None + package_id: int | None = None + + +@attr.s(auto_attribs=True, repr=False) +class CsafProducts: + """List like collection of CSAF products with lookup by ids.""" + + _products: list[CsafProduct] = attr.Factory(list) + _lookup: dict[tuple[int, int | None, int | None, str | None], CsafProduct] = attr.Factory(dict) + + def to_tuples( + self, + attributes: tuple[str, ...], + missing_only: bool = False, + with_id: bool = False, + with_cpe_id: bool = False, + with_pkg_id: bool = False, + ) -> list[tuple[int | str | None, ...]]: + """Transform data to list of tuples with chosen attributes by key. + + :param tuple attributes: Attributes included in the response + :param bool missing_only: Include only products not present in DB (id_=None) + :param bool with_id: Include only products which have product id + :param bool with_cpe_id: Include only products which have cpe_id + :param bool with_pkg_id: Include only products which have either package_name_id or package_id + + Example: + > collection = CsafProducts([CsafProduct(cpe='cpe123', package='kernel', module="module:8", status_id=4)]) + > collection.to_tuples(("cpe", "package", "module")) + > [("cpe123", "kernel", "module:8")] + + """ + res = [] + products: t.Iterable[CsafProduct] = self + if missing_only: + products = filter(lambda x: x.id_ is None, products) + if with_id: + products = filter(lambda x: x.id_, products) + if with_cpe_id: + products = filter(lambda x: x.cpe_id, products) + if with_pkg_id: + products = filter(lambda x: x.package_name_id or x.package_id, products) + + for item in products: + items = [] + for attribute in attributes: + items.append(getattr(item, attribute)) + res.append(tuple(items)) + return res + + def get_by_ids_and_module( + self, + cpe_id: int, + package_name_id: int | None, + package_id: int | None, + module: str | None, + ) -> CsafProduct | None: + """Return product by (cpe_id, package_name_id, package_id, module).""" + key = (cpe_id, package_name_id, package_id, module) + product = self._lookup.get(key) + if product is None: + # not found in _lookup, try to find in _products and add to _lookup + for prod in self: + if (prod.cpe_id, prod.package_name_id, prod.package_id, module) == key: + self.add_to_lookup(prod) + return prod + return product + + def add_to_lookup(self, val: CsafProduct) -> None: + """Add `val` to internal lookup dict.""" + if val.cpe_id is not None: + key = (val.cpe_id, val.package_name_id, val.package_id, val.module) + self._lookup[key] = val + + def append(self, val: CsafProduct) -> None: + """Append `val` to CsafProducts.""" + self._products.append(val) + self.add_to_lookup(val) + + def remove(self, val: CsafProduct) -> None: + """Remove `val` from CsafProducts.""" + if val.cpe_id is not None: + key = (val.cpe_id, val.package_name_id, val.package_id, val.module) + self._lookup.pop(key, None) + self._products.remove(val) + + def __getitem__(self, idx: int) -> CsafProduct: + return self._products[idx] + + def __setitem__(self, idx: int, val: CsafProduct) -> None: + self._products[idx] = val + self.add_to_lookup(val) + + def __iter__(self) -> t.Iterator[CsafProduct]: + return iter(self._products) + + def __next__(self) -> CsafProduct: + return next(iter(self)) + + def __contains__(self, val: CsafProduct) -> bool: + return val in self._products + + def __repr__(self) -> str: + return repr(self._products) + + def __len__(self) -> int: + return len(self._products) + + def __bool__(self) -> bool: + return bool(self._products) @attr.s(auto_attribs=True, repr=False) class CsafCves: """Collection of CSAF CVEs.""" - _cves: dict[str, list[CsafProduct]] = attr.Factory(dict) + _cves: dict[str, CsafProducts] = attr.Factory(dict) - def to_tuples(self, key: str, attributes: tuple[str, ...]) -> list[tuple]: + def to_tuples(self, key: str, attributes: tuple[str, ...]) -> list[tuple[int | str | None, ...]]: """Transform data to list of tuples with chosen attributes by key. Example: @@ -165,15 +281,15 @@ def to_tuples(self, key: str, attributes: tuple[str, ...]) -> list[tuple]: res.append(tuple(items)) return res - @overload - def get(self, key: str, default: None = None) -> Optional[list[CsafProduct]]: + @t.overload + def get(self, key: str, default: None = None) -> CsafProducts | None: ... - @overload - def get(self, key: str, default: list[CsafProduct]) -> list[CsafProduct]: + @t.overload + def get(self, key: str, default: CsafProducts) -> CsafProducts: ... - def get(self, key, default=None): + def get(self, key: str, default: CsafProducts | None = None) -> CsafProducts | None: """Return the value for key if key is in the collection, else default.""" return self._cves.get(key, default) @@ -181,24 +297,24 @@ def update(self, data: CsafCves) -> None: """Update data in collection - same as dict.update().""" self._cves.update(data._cves) # pylint: disable=protected-access - def items(self) -> ItemsView[str, list[CsafProduct]]: + def items(self) -> t.ItemsView[str, CsafProducts]: """Returns CVEs dict key and value pairs.""" return self._cves.items() - def keys(self) -> KeysView: + def keys(self) -> t.KeysView[str]: """Return a list of keys in the _cves dictionary.""" return self._cves.keys() - def __getitem__(self, key: str) -> list[CsafProduct]: + def __getitem__(self, key: str) -> CsafProducts: return self._cves[key] - def __setitem__(self, key: str, val: list[CsafProduct]) -> None: + def __setitem__(self, key: str, val: CsafProducts) -> None: self._cves[key] = val - def __iter__(self) -> Iterator[list[CsafProduct]]: + def __iter__(self) -> t.Iterator[CsafProducts]: return iter(self._cves.values()) - def __next__(self) -> list[CsafProduct]: + def __next__(self) -> CsafProducts: return next(iter(self)) def __contains__(self, key: str) -> bool: @@ -210,6 +326,9 @@ def __repr__(self) -> str: def __len__(self) -> int: return len(self._cves) + def __bool__(self) -> bool: + return bool(self._cves) + @dataclass class CsafData: From dbb0958322e17d97ab455a22487a9b3c816fcde5 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Fri, 8 Mar 2024 19:08:21 +0100 Subject: [PATCH 06/14] fix(csaf): modify csaf_store to reflect new schema --- vmaas/reposcan/database/csaf_store.py | 391 +++++++++++++++++++------- 1 file changed, 293 insertions(+), 98 deletions(-) diff --git a/vmaas/reposcan/database/csaf_store.py b/vmaas/reposcan/database/csaf_store.py index a8547c78a..3746eeb2b 100644 --- a/vmaas/reposcan/database/csaf_store.py +++ b/vmaas/reposcan/database/csaf_store.py @@ -1,46 +1,76 @@ """ Module containing classes for fetching/importing CSAF data from/into database. """ -from collections import defaultdict +import re +from copy import deepcopy +from typing import Any + +from psycopg2 import sql from psycopg2.extras import execute_values +import vmaas.common.rpm_utils as rpm +from vmaas.reposcan.database.cpe_store import CpeStore from vmaas.reposcan.database.object_store import ObjectStore -from vmaas.reposcan.mnm import CSAF_FAILED_DELETE, CSAF_FAILED_IMPORT, CSAF_FAILED_UPDATE -from vmaas.reposcan.redhatcsaf.modeling import CsafCves, CsafData -from vmaas.reposcan.redhatcsaf.modeling import CsafFiles +from vmaas.reposcan.database.package_store import PackageStore +from vmaas.reposcan.mnm import CSAF_FAILED_DELETE +from vmaas.reposcan.mnm import CSAF_FAILED_IMPORT +from vmaas.reposcan.mnm import CSAF_FAILED_UPDATE +from vmaas.reposcan.redhatcsaf import modeling as model + +CVE_PATTERN = re.compile(r"cve-\d+-\d+", re.IGNORECASE) +FAILED_FILE_DELETE = "Failure while deleting CSAF file" +FAILED_FILE_IMPORT = "Failed to import csaf file to DB" +FAILED_PRODUCT_FETCH = "Failure while fetching CSAF products from the DB" +FAILED_PRODUCT_INSERT = "Failure while inserting data into csaf_product" +FAILED_CVE_UPSERT = "Failure while inserting or updating data to csaf_cves" +FAILED_CVE_DELETE = "Failure while removing data from csaf cves" +FAILED_PRODUCT_DELETE = "Failure while deleting data from csaf_product" + + +class CsafStoreException(Exception): + """CsafStoreException exception.""" + + +class CsafStoreSkippedCVE(CsafStoreException): + """CsafStoreSkippedCVE exception.""" class CsafStore(ObjectStore): """Class providing interface for fetching/importing CSAF data from/into the DB.""" - def __init__(self): - super().__init__() - self.csaf_file_map = self._prepare_table_map(cols=("name",), to_cols=("id", "updated"), table="csaf_file") + def __init__(self) -> None: + super().__init__() # type: ignore[no-untyped-call] + self.cpe_store = CpeStore() # type: ignore[no-untyped-call] + self.package_store = PackageStore() # type: ignore[no-untyped-call] + self.csaf_file_map: dict[str, tuple[int, str]] = self._prepare_table_map( # type: ignore[assignment] + cols=("name",), to_cols=("id", "updated"), table="csaf_file" + ) + self.cve2file_id: dict[str, int] = {} - def delete_csaf_file(self, name: str): + def delete_csaf_file(self, name: str) -> None: """Deletes csaf file from DB.""" db_id = self.csaf_file_map[name][0] cur = self.conn.cursor() try: - cur.execute("delete from oval_file where id = %s", (db_id,)) + cur.execute("delete from csaf_file where id = %s", (db_id,)) self.conn.commit() - except Exception: # pylint: disable=broad-except + except Exception as exc: CSAF_FAILED_DELETE.inc() - self.logger.exception("Failed to delete csaf file.") + self.logger.exception("%s: ", FAILED_FILE_DELETE) self.conn.rollback() + raise CsafStoreException(FAILED_FILE_DELETE) from exc finally: cur.close() - def _save_csaf_files(self, csaf_files: CsafFiles) -> list[int]: - res = [] + def _save_csaf_files(self, csaf_files: model.CsafFiles) -> None: cur = self.conn.cursor() try: - res = execute_values( + rows = execute_values( cur, """ insert into csaf_file (name, updated) values %s on conflict (name) do update set updated = excluded.updated - returning id + returning id, name """, csaf_files.to_tuples(("name", "csv_timestamp")), fetch=True, @@ -48,137 +78,302 @@ def _save_csaf_files(self, csaf_files: CsafFiles) -> list[int]: self.conn.commit() except Exception as exc: CSAF_FAILED_IMPORT.inc() - self.logger.exception("Failed to import csaf file to DB: %s", exc) + self.logger.exception("%s: ", FAILED_FILE_IMPORT) + raise CsafStoreException(FAILED_FILE_IMPORT) from exc finally: cur.close() - return [r[0] for r in res] + for row in rows: + # translate file name to cve, e.g. 2023/cve-2023-0030.json -> CVE-2023-0030 + match = re.search(CVE_PATTERN, row[1]) + if match: + cve = match[0].upper() + self.cve2file_id[cve] = row[0] + + def _get_product_attr_id( # type: ignore[return] + self, attr_type: str, mapping: dict[str, int] | dict[tuple[str, ...], int], value: str | tuple[str, ...] + ) -> int: + try: + return mapping[value] # type: ignore[index] + except KeyError as err: + raise KeyError(f"missing {attr_type}={value}") from err + + def _load_product_attr_ids(self, products: model.CsafProducts) -> None: + skipped = [] + for product in products: + try: + product.cpe_id = self._get_product_attr_id("cpe", self.cpe_store.cpe_label_to_id, value=product.cpe) + if product.status_id == model.CsafProductStatus.KNOWN_AFFECTED: + # product for unfixed cve, we have only package_name + product.package_name_id = self._get_product_attr_id( + "package_name", self.package_store.package_name_map, value=product.package + ) + else: + # parse package into NEVRA + try: + name, epoch, ver, rel, arch = rpm.parse_rpm_name(product.package) + name_id = self.package_store.package_name_map[name] + evr_id = self.package_store.evr_map[(epoch, ver, rel)] + arch_id = self.package_store.arch_map[arch] + product.package_id = self._get_product_attr_id( + "package", self.package_store.package_map, value=(name_id, evr_id, arch_id) + ) + except rpm.RPMParseException as err: + self.logger.debug("Skipping product %s, %s", product, err) + skipped.append(product) + continue + + products.add_to_lookup(product) + except (AttributeError, KeyError) as err: + self.logger.debug("Skipping product %s, %s", product, err) + skipped.append(product) - def _get_product_rows(self, products_by_status) -> tuple[dict[str, list[tuple[str, str, str]]], list[tuple[int, str]]]: - to_associate_ids = [] - to_insert = defaultdict(list) + for product in skipped: + products.remove(product) + + def _set_product_ids(self, rows: list[tuple[Any, ...]], products: model.CsafProducts) -> None: + for row in rows: + product = products.get_by_ids_and_module(row[1], row[2], row[3], row[4]) + if product is None: + # log as error, this would be a programming error + self.logger.error("Product %s not found in model.CsafProducts lookup", product) + continue + product.id_ = row[0] + + def _split_product_data(self, products: model.CsafProducts) -> dict[str, dict[str, object]]: + null = sql.SQL("NULL") + not_null = sql.SQL("NOT NULL") + cpe_field = sql.Identifier("cpe_id") + module_field = sql.Identifier("module_stream") + package_name_field = sql.Identifier("package_name_id") + package_field = sql.Identifier("package_id") + + fields_unfixed = [cpe_field, package_name_field] + fields_unfixed_module = [cpe_field, package_name_field, module_field] + fields_fixed = [cpe_field, package_field] + fields_fixed_module = [cpe_field, package_field, module_field] + + unfixed_module = { + "products": [], + "fields": sql.SQL(", ").join(fields_unfixed_module), + "module_null": not_null, + "package_name_null": not_null, + "package_null": null, + } + unfixed_no_module = { + "products": [], + "fields": sql.SQL(", ").join(fields_unfixed), + "module_null": null, + "package_name_null": not_null, + "package_null": null, + } + fixed_module = { + "products": [], + "fields": sql.SQL(", ").join(fields_fixed_module), + "module_null": not_null, + "package_name_null": null, + "package_null": not_null, + } + fixed_no_module = { + "products": [], + "fields": sql.SQL(", ").join(fields_fixed), + "module_null": null, + "package_name_null": null, + "package_null": not_null, + } + + res = { + "unfixed_module": unfixed_module, + "unfixed_no_module": unfixed_no_module, + "fixed_module": fixed_module, + "fixed_no_module": fixed_no_module, + } + for row in products.to_tuples(("cpe_id", "package_name_id", "package_id", "module")): + if row[0] is None: + self.logger.error("Missing cpe_id for product %s", row) + continue + + if any(not isinstance(row[i], (int, str, type(None))) for i in range(4)): + raise TypeError(f"column of product row <{row}> is not of type (int, str, None)") + + if row[1]: # unfixed + if row[3]: # has module + res["unfixed_module"]["products"].append((row[0], row[1], row[3])) # type: ignore[attr-defined] + else: + res["unfixed_no_module"]["products"].append((row[0], row[1])) # type: ignore[attr-defined] + elif row[2]: # fixed + if row[3]: # has module + res["fixed_module"]["products"].append((row[0], row[2], row[3])) # type: ignore[attr-defined] + else: + res["fixed_no_module"]["products"].append((row[0], row[2])) # type: ignore[attr-defined] + return res + + def _update_product_ids(self, products: model.CsafProducts) -> None: + if not products: + return + + all_rows = [] + query = sql.SQL( + """ + SELECT id, cpe_id, package_name_id, package_id, module_stream + FROM csaf_product + WHERE ({fields}) in %s + AND module_stream IS {module_null} + AND package_name_id IS {package_name_null} + AND package_id IS {package_null} + """ + ) + cur = self.conn.cursor() try: - cur = self.conn.cursor() - for status, products in products_by_status.items(): - if not products: - continue - cur.execute("""SELECT id, cpe, package, module - FROM csaf_products - WHERE ((cpe, package, module) in %s AND module IS NOT NULL) - OR ((cpe, package) in %s AND module IS NULL)""", - (tuple(products), tuple((cpe, package) for cpe, package, _ in products))) - rows = cur.fetchall() - for row in rows: - to_associate_ids.append((row[0], status)) - for product in products: - insert = True - for row in rows: - if product == (row[1], row[2], row[3]): - insert = False - break - if insert: - to_insert[status].append(product) + self._load_product_attr_ids(products) + + for key, val in self._split_product_data(products).items(): + self.logger.debug("loading <%s> products", key) + product_tuples: tuple[tuple[int | str, ...]] = tuple(val["products"]) # type: ignore + if product_tuples: + formatted_query = query.format( + fields=val["fields"], # type: ignore[arg-type] + module_null=val["module_null"], # type: ignore[arg-type] + package_name_null=val["package_name_null"], # type: ignore[arg-type] + package_null=val["package_null"], # type: ignore[arg-type] + ) + cur.execute(formatted_query, (product_tuples,)) + rows = cur.fetchall() + all_rows.extend(rows) + self._set_product_ids(all_rows, products) except Exception as exc: CSAF_FAILED_IMPORT.inc() - self.logger.exception("Failure while fetching CSAF products from the DB: %s", exc) + self.logger.exception("%s: ", FAILED_PRODUCT_FETCH) + raise CsafStoreException(FAILED_PRODUCT_FETCH) from exc finally: cur.close() - return to_insert, to_associate_ids + def _insert_missing_products(self, products: model.CsafProducts) -> None: + if not products: + return - def _insert_missing_products(self, products_by_status: dict) -> list[tuple[int, str]]: - inserted_products = [] + cur = self.conn.cursor() try: - cur = self.conn.cursor() - for status, products in products_by_status.items(): - if products: - ids = execute_values(cur, """ - INSERT INTO csaf_products (cpe, package, module) - VALUES %s returning id; - """, (products), fetch=True) - inserted_products.extend([(result[0], status) for result in ids]) - self.conn.commit() + rows = execute_values( + cur, + """ + INSERT INTO csaf_product (cpe_id, package_name_id, package_id, module_stream) + VALUES %s + RETURNING id, cpe_id, package_name_id, package_id, module_stream + """, + products.to_tuples( + ("cpe_id", "package_name_id", "package_id", "module"), + missing_only=True, + with_cpe_id=True, + with_pkg_id=True, + ), + fetch=True, + ) + self.conn.commit() + self._set_product_ids(rows, products) except Exception as exc: CSAF_FAILED_IMPORT.inc() - self.logger.exception("Failure while inserting data into csaf_products: %s", exc) + self.logger.exception("%s: ", FAILED_PRODUCT_INSERT) self.conn.rollback() + raise CsafStoreException(FAILED_PRODUCT_INSERT) from exc finally: cur.close() - return inserted_products + def _insert_cves(self, cve: str, products: model.CsafProducts) -> None: + if not products: + raise CsafStoreSkippedCVE("cannot map any products") - def _insert_cves(self, cve, products): + cur = self.conn.cursor() try: - cur = self.conn.cursor() - if products: - execute_values(cur, """ - INSERT INTO csaf_cves (cve, csaf_product_id, product_status_id) + cur.execute("SELECT id FROM cve WHERE UPPER(name) = %s", (cve,)) + if (row := cur.fetchone()) is None: + raise CsafStoreException(f"{cve} not found in DB") + + file_id = self.cve2file_id[cve] + to_upsert = [ + (row[0], id_, status_id, file_id) + for id_, status_id in products.to_tuples(("id_", "status_id"), with_id=True) + ] + execute_values( + cur, + """ + INSERT INTO csaf_cve_product (cve_id, csaf_product_id, csaf_product_status_id, csaf_file_id) VALUES %s - ON CONFLICT (cve, csaf_product_id) - DO UPDATE SET product_status_id = EXCLUDED.product_status_id; - """, [(cve, product[0], product[1]) for product in products]) - self.conn.commit() + ON CONFLICT (cve_id, csaf_product_id) + DO UPDATE SET csaf_product_status_id = EXCLUDED.csaf_product_status_id + """, + to_upsert, + ) + self.conn.commit() except Exception as exc: CSAF_FAILED_IMPORT.inc() CSAF_FAILED_UPDATE.inc() - self.logger.exception("Failure while inserting or updating data to csaf_cves: %s", exc) + self.logger.exception("%s: ", FAILED_CVE_UPSERT) self.conn.rollback() + raise CsafStoreException(FAILED_CVE_UPSERT) from exc finally: cur.close() - def _remove_cves(self, cve, ids): - if not ids: + def _remove_cves(self, cve: str, products: model.CsafProducts) -> None: + if not products: return + + cur = self.conn.cursor() try: - cur = self.conn.cursor() - cur.execute("""DELETE FROM csaf_cves WHERE cve = %s and csaf_product_id not in %s""", - (cve, tuple(row[1] for row in ids))) + cur.execute( + """ + DELETE FROM csaf_cve_product ccp + USING cve + WHERE ccp.cve_id = cve.id + AND cve.name = %s + AND csaf_product_id NOT IN %s + """, + (cve, tuple(x.id_ for x in products)), + ) self.conn.commit() except Exception as exc: CSAF_FAILED_DELETE.inc() - self.logger.exception("Failure while removing data from csaf cves: %s", exc) + self.logger.exception("%s: ", FAILED_CVE_DELETE) self.conn.rollback() + raise CsafStoreException(FAILED_CVE_DELETE) from exc finally: cur.close() - def _map_products_by_status(self, products) -> dict[int, list[tuple[str, str, str]]]: - products_by_status = defaultdict(list) + def _populate_cves(self, csaf_cves: model.CsafCves) -> None: + for cve, products in csaf_cves.items(): + products_copy = deepcopy(products) # only for logging of failed cves + try: + self._update_product_ids(products) + self._insert_missing_products(products) + self._remove_cves(cve, products) + self._insert_cves(cve, products) + except CsafStoreSkippedCVE as exc: + self.logger.warning("Skipping cve: %s reason: %s", cve, exc) + self.logger.debug("Skipping cve: %s products: %s reason: %s", cve, products_copy, exc) + except CsafStoreException: + self.logger.exception("Failed to populate cve: %s products: %s ", cve, products_copy) - for product in products: - status_id = product[2] - product_data = (product[0], product[1], product[3]) - products_by_status[status_id].append(product_data) - - return products_by_status - - def _populate_cves(self, csaf_cves: CsafCves): - for cve in csaf_cves.keys(): - cve_products = csaf_cves.to_tuples(cve, ("cpe", "package", "status_id", "module")) - products = self._map_products_by_status(cve_products) - to_insert, to_associate_ids = self._get_product_rows(products) - inserted_ids = self._insert_missing_products(to_insert) - all_ids = to_associate_ids+inserted_ids - self._remove_cves(cve, all_ids) - self._insert_cves(cve, all_ids) - - def _delete_unreferenced_products(self): + def _delete_unreferenced_products(self) -> None: + cur = self.conn.cursor() try: - cur = self.conn.cursor() - cur.execute("""DELETE FROM csaf_products + cur.execute( + """DELETE FROM csaf_product WHERE id IN ( SELECT p.id - FROM csaf_products AS p - LEFT JOIN csaf_cves AS c ON c.csaf_product_id = p.id - WHERE c.csaf_product_id IS NULL)""") + FROM csaf_product AS p + LEFT JOIN csaf_cve_product AS c ON c.csaf_product_id = p.id + WHERE c.csaf_product_id IS NULL)""" + ) self.conn.commit() except Exception as exc: CSAF_FAILED_DELETE.inc() - self.logger.exception("Failure while deleting data from csaf_products: %s", exc) + self.logger.exception("%s: ", FAILED_PRODUCT_DELETE) self.conn.rollback() + raise CsafStoreException(FAILED_PRODUCT_DELETE) from exc finally: cur.close() - def store(self, csaf_data: CsafData): + def store(self, csaf_data: model.CsafData) -> None: """Store collection of CSAF files into DB.""" self._save_csaf_files(csaf_data.files) self._populate_cves(csaf_data.cves) From 97814476e414d04db89bcb3c9786fbd1d7cf151c Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Mon, 11 Mar 2024 21:19:59 +0100 Subject: [PATCH 07/14] chore(mypy): improve checks and fix issues found by mypy --- .github/workflows/tests.yml | 2 +- pyproject.toml | 14 +- vmaas/common/batch_list.py | 13 +- vmaas/common/config.py | 8 +- vmaas/common/date_utils.py | 2 +- vmaas/common/logging_utils.py | 2 +- vmaas/common/rpm_utils.py | 3 +- vmaas/reposcan/database/object_store.py | 7 +- vmaas/reposcan/download/downloader.py | 10 +- vmaas/reposcan/redhatcsaf/csaf_controller.py | 36 ++-- .../redhatcsaf/test/test_controller.py | 17 +- .../reposcan/redhatcsaf/test/test_modeling.py | 186 ++++++++++-------- 12 files changed, 168 insertions(+), 132 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 49f5a7943..5a4f2b4dd 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -76,4 +76,4 @@ jobs: flake8 vmaas/ - name: Run mypy run: | - mypy . + mypy diff --git a/pyproject.toml b/pyproject.toml index 4a6ff9368..73c50889b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,7 +54,13 @@ requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" [tool.mypy] -ignore_missing_imports = true -exclude = '''(?x)( - ^vmaas/webapp # exclude vmaas/webapp dir -)''' +disallow_any_generics = true +disallow_untyped_defs = true +disallow_untyped_calls = true +warn_return_any = true +warn_unreachable = true +follow_imports = "silent" +packages = """ + vmaas.reposcan.database.csaf_store, + vmaas.reposcan.redhatcsaf +""" diff --git a/vmaas/common/batch_list.py b/vmaas/common/batch_list.py index 44a06708c..6befa3931 100644 --- a/vmaas/common/batch_list.py +++ b/vmaas/common/batch_list.py @@ -2,6 +2,7 @@ Module containing class for list of batches. """ import os +import typing as t BATCH_MAX_SIZE = int(os.getenv('BATCH_MAX_SIZE', "500")) BATCH_MAX_FILESIZE = int(os.getenv('BATCH_MAX_FILESIZE', "14_000_000_000")) @@ -10,18 +11,18 @@ class BatchList: """List of lists with defined maximum size of inner lists or by arbitrary file_size of each item""" - def __init__(self): + def __init__(self) -> None: self.batches = [] self.last_batch_filesize = 0 - def __iter__(self): + def __iter__(self) -> t.Iterator[list]: return iter(self.batches) - def clear(self): + def clear(self) -> None: """Clear all previously added items.""" self.batches = [] - def add_item(self, item, file_size=0): + def add_item(self, item: t.Any, file_size: int = 0) -> None: """Add item into the last batch. Create new batch if there is no batch or last batch is full.""" if self.batches: last_batch = self.batches[-1] @@ -41,9 +42,9 @@ def add_item(self, item, file_size=0): last_batch.append(item) self.last_batch_filesize += file_size - def __len__(self): + def __len__(self) -> int: return len(self.batches) - def get_total_items(self): + def get_total_items(self) -> int: """Return total item count in all batches.""" return sum(len(batch) for batch in self.batches) diff --git a/vmaas/common/config.py b/vmaas/common/config.py index b1345717a..de5b80e39 100644 --- a/vmaas/common/config.py +++ b/vmaas/common/config.py @@ -23,7 +23,7 @@ def __call__(cls, *args, **kwargs): class BaseConfig: """Base configuration, same for clowder and non-clowder.""" - def __init__(self): + def __init__(self) -> None: self.postgresql_writer_password = os.getenv( "POSTGRESQL_WRITER_PASSWORD", "vmaas_writer_pwd" ) @@ -49,7 +49,7 @@ def __init__(self): class Config(BaseConfig, metaclass=Singleton): """VMaaS configuration singleton.""" - def __init__(self): + def __init__(self) -> None: if not app_common_python.isClowderEnabled(): raise EnvironmentError("Missing Clowder config.") @@ -57,7 +57,7 @@ def __init__(self): self._cfg = app_common_python.LoadedConfig self.clowder() - def clowder(self): + def clowder(self) -> None: """Configuration from Clowder.""" self.db_name = getattr(self._cfg.database, "name", "") self.db_user = getattr(self._cfg.database, "username", "") @@ -86,7 +86,7 @@ def clowder(self): self.private_port = self._cfg.privatePort self.metrics_port = self._cfg.metricsPort - def _build_url(self, endpoint, scheme="http"): + def _build_url(self, endpoint: object, scheme="http") -> str: scheme = f"{scheme}s" if self.tls_ca_path else scheme port = endpoint.tlsPort if self.tls_ca_path else endpoint.port return f"{scheme}://{endpoint.hostname}:{port}" diff --git a/vmaas/common/date_utils.py b/vmaas/common/date_utils.py index 072424234..8256422bc 100644 --- a/vmaas/common/date_utils.py +++ b/vmaas/common/date_utils.py @@ -7,7 +7,7 @@ from dateutil import tz as dateutil_tz -def parse_datetime(date): +def parse_datetime(date: str) -> datetime: """Parse date from string in ISO format.""" if date is None: return None diff --git a/vmaas/common/logging_utils.py b/vmaas/common/logging_utils.py index 84c2588ce..43eb71ea0 100644 --- a/vmaas/common/logging_utils.py +++ b/vmaas/common/logging_utils.py @@ -133,7 +133,7 @@ def init_logging(num_servers=1): setup_cw_logging(logger) -def get_logger(name): +def get_logger(name: str) -> logging.Logger: """ Set logging level and return logger. Don't set custom logging level in root handler to not display debug messages from underlying libraries. diff --git a/vmaas/common/rpm_utils.py b/vmaas/common/rpm_utils.py index 86ef0f748..2628017de 100644 --- a/vmaas/common/rpm_utils.py +++ b/vmaas/common/rpm_utils.py @@ -21,7 +21,8 @@ class RPMParseException(Exception): r'((?P[0-9]+):)?(?P[^:]+)(?(e1)-|-((?P[0-9]+):)?)(?P[^-:]+)-(?P[^-:]+)\.(?P[a-z0-9_]+)') -def parse_rpm_name(rpm_name, default_epoch=None, raise_exception=False): +def parse_rpm_name(rpm_name: str, default_epoch: str | None = None, + raise_exception: bool = False) -> tuple[str, str, str, str, str]: """ Extract components from rpm name. """ diff --git a/vmaas/reposcan/database/object_store.py b/vmaas/reposcan/database/object_store.py index cd8630408..226974d98 100644 --- a/vmaas/reposcan/database/object_store.py +++ b/vmaas/reposcan/database/object_store.py @@ -1,10 +1,12 @@ """ Module containing shared code between various *Store classes """ - +import typing as t from vmaas.common.logging_utils import get_logger from vmaas.reposcan.database.database_handler import DatabaseHandler +PrepareTableMapT = dict[str, int] | dict[str, tuple[t.Any, ...]] | dict[tuple[str, ...], int] | dict[tuple[str, ...], tuple[t.Any, ...]] + class ObjectStore: """Shared code between various *Store classes""" @@ -42,7 +44,8 @@ def _get_modules_in_repo(self, repo_id): cur.close() return modules_in_repo - def _prepare_table_map(self, cols, table, to_cols=None, where=None): + def _prepare_table_map(self, cols: t.Iterable[str], table: str, to_cols: t.Iterable[str] | None = None, + where: str | None = None) -> PrepareTableMapT: """Create map from table map[columns] -> or(column, tuple(columns)).""" if not to_cols: to_cols = ["id"] diff --git a/vmaas/reposcan/download/downloader.py b/vmaas/reposcan/download/downloader.py index 32c5ec38d..45ce58010 100644 --- a/vmaas/reposcan/download/downloader.py +++ b/vmaas/reposcan/download/downloader.py @@ -4,6 +4,7 @@ import os from threading import Thread from queue import Queue, Empty +from pathlib import Path from urllib3.exceptions import ProtocolError @@ -27,7 +28,8 @@ class DownloadItem: and result HTTP status code of the download operation. """ - def __init__(self, source_url=None, target_path=None, ca_cert=None, cert=None, key=None): + def __init__(self, source_url: str | None = None, target_path: Path | None = None, + ca_cert: str | None = None, cert: str | None = None, key: str | None = None) -> None: self.source_url = source_url self.target_path = target_path self.ca_cert = ca_cert @@ -124,16 +126,16 @@ class FileDownloader: are finished. """ - def __init__(self): + def __init__(self) -> None: self.queue = Queue() self.logger = get_logger(__name__) self.num_threads = int(os.getenv('THREADS', DEFAULT_THREADS)) - def add(self, download_item): + def add(self, download_item: DownloadItem) -> None: """Add DownloadItem object into the queue.""" self.queue.put(download_item) - def run(self, headers_only=False): + def run(self, headers_only: bool = False) -> None: """Start processing download queue using multiple threads.""" progress_logger = ProgressLogger(self.logger, self.queue.qsize()) self.logger.info("Downloading started.") diff --git a/vmaas/reposcan/redhatcsaf/csaf_controller.py b/vmaas/reposcan/redhatcsaf/csaf_controller.py index a889ccf0f..57ed96090 100644 --- a/vmaas/reposcan/redhatcsaf/csaf_controller.py +++ b/vmaas/reposcan/redhatcsaf/csaf_controller.py @@ -5,6 +5,7 @@ import os import shutil import tempfile +import typing as t from pathlib import Path from vmaas.common.batch_list import BatchList @@ -32,29 +33,32 @@ class CsafController: """Class for importing/syncing set of CSAF files into DB.""" - def __init__(self): + def __init__(self) -> None: self.logger = get_logger(__name__) self.downloader = FileDownloader() self.downloader.num_threads = 1 # rh.com returns 403 when downloading too quickly (DDoS protection?) self.csaf_store = CsafStore() - self.tmp_directory = Path(tempfile.mkdtemp(prefix="csaf-")) + self.tmp_directory: Path | None = Path(tempfile.mkdtemp(prefix="csaf-")) self.index_path = self.tmp_directory / CSAF_VEX_INDEX_CSV self.cfg = Config() - def _download_index(self) -> dict[str, int]: + def _download_index(self) -> dict[Path, int]: """Download CSAF index changes.csv file.""" item = DownloadItem(source_url=CSAF_VEX_BASE_URL + CSAF_VEX_INDEX_CSV, target_path=self.index_path) self.downloader.add(item) self.downloader.run() # return failed download - if item.status_code not in VALID_HTTP_CODES: + if item.status_code not in VALID_HTTP_CODES and item.target_path: return {item.target_path: item.status_code} return {} - def _download_csaf_files(self, batch) -> dict[str, int]: + def _download_csaf_files(self, batch: list[CsafFile]) -> dict[Path, int]: """Download CSAF files.""" download_items = [] for csaf_file in batch: + if not self.tmp_directory: + self.logger.error("Missing temporary directory for csaf download") + return {} local_path = self.tmp_directory / csaf_file.name os.makedirs(os.path.dirname(local_path), exist_ok=True) # Make sure subdirs exist item = DownloadItem(source_url=CSAF_VEX_BASE_URL + csaf_file.name, target_path=local_path) @@ -64,16 +68,18 @@ def _download_csaf_files(self, batch) -> dict[str, int]: self.downloader.run() # Return failed downloads return { - item.target_path: item.status_code for item in download_items if item.status_code not in VALID_HTTP_CODES + item.target_path: item.status_code + for item in download_items + if item.status_code not in VALID_HTTP_CODES and item.target_path } - def clean(self): + def clean(self) -> None: """Clean downloaded files for given batch.""" if self.tmp_directory: shutil.rmtree(self.tmp_directory) self.tmp_directory = None - def store(self): + def store(self) -> None: """Process and store CSAF objects to DB.""" self.logger.info("Checking CSAF index.") failed = self._download_index() @@ -86,8 +92,8 @@ def store(self): db_csaf_files = self.csaf_store.csaf_file_map.copy() batches = BatchList() - csaf_files = CsafFiles.from_table_map_and_csv(db_csaf_files, self.index_path) - files_to_sync = csaf_files + csaf_files = CsafFiles.from_table_map_and_csv(db_csaf_files, self.index_path) # type: ignore[arg-type] + files_to_sync: t.Iterable[CsafFile] = csaf_files if not CSAF_SYNC_ALL_FILES: files_to_sync = csaf_files.out_of_date @@ -120,6 +126,10 @@ def parse_csaf_file(self, csaf_file: CsafFile) -> CsafCves: product_cpe = {} cves = CsafCves() + if not self.tmp_directory: + self.logger.error("Missing temporary directory for csaf files") + raise FileNotFoundError("Missing csaf tmp dir") + with open(self.tmp_directory / csaf_file.name, "r", encoding="utf-8") as csaf_json: csaf = json.load(csaf_json) product_cpe = self._parse_product_tree(csaf) @@ -127,7 +137,7 @@ def parse_csaf_file(self, csaf_file: CsafFile) -> CsafCves: cves.update(unfixed_cves) return cves - def _parse_vulnerabilities(self, csaf: dict, product_cpe: dict) -> CsafCves: + def _parse_vulnerabilities(self, csaf: dict[str, t.Any], product_cpe: dict[str, str]) -> CsafCves: # parse only CVEs with `known_affected` products aka `unfixed` CVEs if any(x != "known_affected" for x in self.cfg.csaf_product_status_list): raise NotImplementedError("parsing of csaf products other than 'known_affected' not supported") @@ -155,14 +165,14 @@ def _parse_vulnerabilities(self, csaf: dict, product_cpe: dict) -> CsafCves: return unfixed_cves - def _parse_product_tree(self, csaf: dict) -> dict[str, str]: + def _parse_product_tree(self, csaf: dict[str, t.Any]) -> dict[str, str]: product_cpe: dict[str, str] = {} for branches in csaf.get("product_tree", {}).get("branches", []): self._parse_branches(branches, product_cpe) return product_cpe - def _parse_branches(self, branches: dict, product_cpe: dict[str, str]): + def _parse_branches(self, branches: dict[str, t.Any], product_cpe: dict[str, str]) -> None: if branches.get("category") not in ("vendor", "product_family"): return sub_branches = branches.get("branches", []) diff --git a/vmaas/reposcan/redhatcsaf/test/test_controller.py b/vmaas/reposcan/redhatcsaf/test/test_controller.py index 34ccc6b9b..98dd6af46 100644 --- a/vmaas/reposcan/redhatcsaf/test/test_controller.py +++ b/vmaas/reposcan/redhatcsaf/test/test_controller.py @@ -3,37 +3,36 @@ from datetime import datetime import pytest +from psycopg2.extensions import connection +import vmaas.reposcan.redhatcsaf.modeling as m from vmaas.reposcan.conftest import EXPECTED_CSAF from vmaas.reposcan.redhatcsaf.csaf_controller import CsafController -from vmaas.reposcan.redhatcsaf.modeling import CsafData -from vmaas.reposcan.redhatcsaf.modeling import CsafFile -from vmaas.reposcan.redhatcsaf.modeling import CsafFiles class TestCsafController: """CsafController tests.""" @pytest.fixture - def csaf(self, db_conn): # pylint: disable=unused-argument + def csaf(self, db_conn: connection) -> CsafController: # pylint: disable=unused-argument """Fixture returning CsafController obj with tmp_directory se to current directory.""" csaf = CsafController() csaf.tmp_directory = pathlib.Path(__file__).parent.resolve() return csaf @pytest.mark.parametrize("data", EXPECTED_CSAF, ids=[x[0] for x in EXPECTED_CSAF]) - def test_parse_csaf_file(self, data, csaf): + def test_parse_csaf_file(self, data: tuple[str, m.CsafCves], csaf: CsafController) -> None: """Test CSAF JSON file parsing.""" csaf_json, expected = data - csaf_file = CsafFile(csaf_json, None) + csaf_file = m.CsafFile(csaf_json, datetime.now()) parsed = csaf.parse_csaf_file(csaf_file) assert parsed == expected @pytest.mark.parametrize("data", EXPECTED_CSAF, ids=[x[0] for x in EXPECTED_CSAF]) - def test_csaf_store(self, data, csaf: CsafController): + def test_csaf_store(self, data: tuple[str, m.CsafCves], csaf: CsafController) -> None: """Parse CSAF JSON and store its data to DB""" csaf_json, _ = data - csaf_file = CsafFile(csaf_json, datetime(2022, 2, 14, 15, 30)) + csaf_file = m.CsafFile(csaf_json, datetime(2022, 2, 14, 15, 30)) csaf_cves = csaf.parse_csaf_file(csaf_file) - csaf_data = CsafData(CsafFiles({csaf_json: csaf_file}), csaf_cves) + csaf_data = m.CsafData(m.CsafFiles({csaf_json: csaf_file}), csaf_cves) csaf.csaf_store.store(csaf_data) diff --git a/vmaas/reposcan/redhatcsaf/test/test_modeling.py b/vmaas/reposcan/redhatcsaf/test/test_modeling.py index d89019d79..3cb3a9d17 100644 --- a/vmaas/reposcan/redhatcsaf/test/test_modeling.py +++ b/vmaas/reposcan/redhatcsaf/test/test_modeling.py @@ -1,12 +1,15 @@ """Unit tests of CSAF models.""" from collections.abc import Iterable +from pathlib import Path from datetime import datetime from datetime import timedelta -from typing import KeysView +from typing import Any import pytest -import vmaas.reposcan.redhatcsaf.modeling as model +import vmaas.reposcan.redhatcsaf.modeling as m + +DictCollection = m.CsafCves | m.CsafFiles # pylint: disable=too-many-public-methods @@ -14,64 +17,66 @@ class TestModels: """Test CSAF models.""" @pytest.fixture - def csaf_product(self): + def csaf_product(self) -> m.CsafProduct: """Returns CSAF product.""" - return model.CsafProduct("cpe", "package", 4, "module") + return m.CsafProduct("cpe", "package", 4, "module") @pytest.fixture - def csaf_product_list(self, csaf_product): + def csaf_product_list(self, csaf_product: m.CsafProduct) -> m.CsafProducts: """Returns list of CSAF products.""" - return [csaf_product] + return m.CsafProducts([csaf_product]) @pytest.fixture - def csaf_file(self): + def csaf_file(self) -> m.CsafFile: """Returns CSAF file.""" now = datetime.now() now1d = now + timedelta(days=1) - return model.CsafFile("name", now1d, now) + return m.CsafFile("name", now1d, now) @pytest.fixture - def csaf_cves(self, csaf_product_list): + def csaf_cves(self, csaf_product_list: m.CsafProducts) -> m.CsafCves: """Returns CSAF CVEs collection.""" - return model.CsafCves({"key1": csaf_product_list, "key2": csaf_product_list}) + return m.CsafCves({"key1": csaf_product_list, "key2": csaf_product_list}) @pytest.fixture - def csaf_files(self, csaf_file): + def csaf_files(self, csaf_file: m.CsafFile) -> m.CsafFiles: """Returns CSAF files collection.""" - return model.CsafFiles({"key1": csaf_file, "key2": csaf_file}) + return m.CsafFiles({"key1": csaf_file, "key2": csaf_file}) @pytest.fixture - def csaf_product_updated(self): + def csaf_product_updated(self) -> m.CsafProduct: """Returns CSAF product.""" - return model.CsafProduct("cpe_updated", "package_updated", "module_updated") + return m.CsafProduct("cpe_updated", "package_updated", 4, "module_updated") @pytest.fixture - def csaf_product_list_updated(self, csaf_product_updated): + def csaf_product_list_updated(self, csaf_product_updated: m.CsafProduct) -> m.CsafProducts: """Returns list of CSAF products.""" - return [csaf_product_updated] + return m.CsafProducts([csaf_product_updated]) @pytest.fixture - def csaf_file_updated(self): + def csaf_file_updated(self) -> m.CsafFile: """Returns CSAF file.""" now = datetime.now() now1d = now + timedelta(days=1) - return model.CsafFile("name_updated", now1d, now) + return m.CsafFile("name_updated", now1d, now) @pytest.fixture - def csaf_cves_updated(self, csaf_product_list_updated): + def csaf_cves_updated(self, csaf_product_list_updated: m.CsafProducts) -> m.CsafCves: """Returns CSAF CVEs collection.""" - return model.CsafCves({"key2": csaf_product_list_updated}) + return m.CsafCves({"key2": csaf_product_list_updated}) @pytest.fixture - def csaf_files_updated(self, csaf_file_updated): + def csaf_files_updated(self, csaf_file_updated: m.CsafFile) -> m.CsafFiles: """Returns CSAF files collection.""" - return model.CsafFiles({"key2": csaf_file_updated}) + return m.CsafFiles({"key2": csaf_file_updated}) - @pytest.mark.parametrize("collection, expected", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file"))) - def test_get(self, collection, expected, request): + @pytest.mark.parametrize( + "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + ) + def test_get(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test get()""" - collection = request.getfixturevalue(collection) - expected = request.getfixturevalue(expected) + collection = request.getfixturevalue(collection_name) + expected = request.getfixturevalue(expected_arg) res = collection.get("key1") assert res == expected @@ -82,70 +87,74 @@ def test_get(self, collection, expected, request): res = collection.get("key99", "default") assert res == "default" - @pytest.mark.parametrize("collection, expected", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file"))) - def test_getitem(self, collection, expected, request): + @pytest.mark.parametrize( + "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + ) + def test_getitem(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test __getitem__.""" - collection = request.getfixturevalue(collection) + collection: DictCollection = request.getfixturevalue(collection_name) res = collection["key1"] - assert res == request.getfixturevalue(expected) + assert res == request.getfixturevalue(expected_arg) with pytest.raises(KeyError): collection["key99"] # pylint: disable=pointless-statement @pytest.mark.parametrize( - "collection, expected", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) + "collection_name, expected_arg", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) ) - def test_setitem(self, collection, expected, request): + def test_setitem(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test __setitem__""" - collection = request.getfixturevalue(collection) - expected = request.getfixturevalue(expected) + collection: DictCollection = request.getfixturevalue(collection_name) + expected = request.getfixturevalue(expected_arg) collection["key2"] = expected["key2"] assert collection["key2"] == expected["key2"] @pytest.mark.parametrize( - "collection, expected", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) + "collection_name, expected_arg", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) ) - def test_update(self, collection, expected, request): + def test_update(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test update()""" - collection = request.getfixturevalue(collection) - expected = request.getfixturevalue(expected) + collection: DictCollection = request.getfixturevalue(collection_name) + expected = request.getfixturevalue(expected_arg) collection.update(expected) assert collection["key2"] == expected["key2"] - @pytest.mark.parametrize("collection", ("csaf_cves", "csaf_files")) - def test_iter(self, collection, request): + @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files")) + def test_iter(self, collection_name: str, request: pytest.FixtureRequest) -> None: """Test __iter__""" - collection = request.getfixturevalue(collection) + collection: DictCollection = request.getfixturevalue(collection_name) assert isinstance(collection, Iterable) - @pytest.mark.parametrize("collection, expected", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file"))) - def test_next(self, collection, expected, request): + @pytest.mark.parametrize( + "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + ) + def test_next(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test __next__""" - collection = request.getfixturevalue(collection) - expected = request.getfixturevalue(expected) + collection: DictCollection = request.getfixturevalue(collection_name) + expected = request.getfixturevalue(expected_arg) assert next(collection) == expected - @pytest.mark.parametrize("collection", ("csaf_cves", "csaf_files")) - def test_contains(self, collection, request): + @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files")) + def test_contains(self, collection_name: str, request: pytest.FixtureRequest) -> None: """Test __contains__""" - collection = request.getfixturevalue(collection) + collection: DictCollection = request.getfixturevalue(collection_name) assert "key1" in collection assert "key99" not in collection @pytest.mark.parametrize( "obj, expected", (("csaf_cves", "key1"), ("csaf_files", "key1"), ("csaf_product", "cpe"), ("csaf_file", "name")) ) - def test_repr(self, obj, expected, request): + def test_repr(self, obj: str, expected: str, request: pytest.FixtureRequest) -> None: """Test __repr__""" obj = request.getfixturevalue(obj) res = repr(obj) assert isinstance(res, str) assert expected in res - @pytest.mark.parametrize("collection", ("csaf_cves", "csaf_files")) - def test_len(self, collection, request): + @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files")) + def test_len(self, collection_name: str, request: pytest.FixtureRequest) -> None: """Test __len__""" - collection = request.getfixturevalue(collection) + collection: DictCollection = request.getfixturevalue(collection_name) res = len(collection) assert res == 2 @@ -154,29 +163,29 @@ def test_len(self, collection, request): ( ("CsafFile", ("x", "y")), ("CsafFiles", None), - ("CsafProduct", ("x", "y", "z")), + ("CsafProduct", ("x", "y", 4, "z")), ("CsafCves", None), ("CsafData", None), ), ) - def test_instantiate(self, class_name, args): + def test_instantiate(self, class_name: str, args: tuple[str | int] | None) -> None: """Test class instantiation""" - class_ = getattr(model, class_name) + class_ = getattr(m, class_name) if args: class_(*args) else: class_() - def test_csaf_file_out_of_date(self, csaf_file): + def test_csaf_file_out_of_date(self, csaf_file: m.CsafFile) -> None: """Test CsafFile.out_of_date""" assert csaf_file.out_of_date - def test_csaf_files_out_of_date(self, csaf_file): + def test_csaf_files_out_of_date(self, csaf_file: m.CsafFile) -> None: """Test CsafFiles.out_of_date""" - collection = model.CsafFiles({"x": csaf_file, "y": csaf_file}) + collection = m.CsafFiles({"x": csaf_file, "y": csaf_file}) assert len(list(collection.out_of_date)) == 2 - def test_from_table_map_and_csv(self, tmp_path): + def test_from_table_map_and_csv(self, tmp_path: Path) -> None: """Test CsafFiles.from_table_map_and_csv""" now = datetime.now() csv_file = tmp_path / "test_csaf" / "test.csv" @@ -187,8 +196,9 @@ def test_from_table_map_and_csv(self, tmp_path): modified = datetime.now() csv_file.write_text(f"file2,{str(modified)}\r\nfile3,{str(modified)}") - collection = model.CsafFiles.from_table_map_and_csv(table_map, csv_file) + collection = m.CsafFiles.from_table_map_and_csv(table_map, csv_file) assert collection["file1"].csv_timestamp == collection["file1"].db_timestamp == now + assert collection["file2"].db_timestamp is not None assert collection["file2"].csv_timestamp > collection["file2"].db_timestamp assert collection["file2"].csv_timestamp == modified assert collection["file3"].csv_timestamp == modified @@ -198,26 +208,28 @@ def test_from_table_map_and_csv(self, tmp_path): assert len(out_of_date) == 2 @pytest.mark.parametrize( - "collection, attr_tuple, by_key", + "collection_name, attr_tuple, by_key", (("csaf_cves", ("cpe", "package", "module"), "key1"), ("csaf_files", ("name",), None)), ) - def test_to_tuples(self, collection, attr_tuple, by_key, request): + def test_to_tuples( + self, collection_name: str, attr_tuple: tuple[str], by_key: str | None, request: pytest.FixtureRequest + ) -> None: """Test collection.to_tuples()""" - collection = request.getfixturevalue(collection) + collection: DictCollection = request.getfixturevalue(collection_name) - def _assert_tuples(item): + def _assert_tuples(item: tuple[Any, ...]) -> None: assert len(item) == len(attr_tuple) for attr in attr_tuple: assert attr in item - if by_key: - res, *_ = collection.to_tuples(by_key, attr_tuple) - _assert_tuples(res) - else: - res, *_ = collection.to_tuples(attr_tuple) - _assert_tuples(res) + if by_key and isinstance(collection, m.CsafCves): + cves_tuple, *_ = collection.to_tuples(by_key, attr_tuple) + _assert_tuples(cves_tuple) + elif isinstance(collection, m.CsafFiles): + files_tuple, *_ = collection.to_tuples(attr_tuple) + _assert_tuples(files_tuple) - def test_to_tuples_exception(self, csaf_cves, csaf_files): + def test_to_tuples_exception(self, csaf_cves: m.CsafCves, csaf_files: m.CsafFiles) -> None: """Test exceptions from collection.to_tuples()""" with pytest.raises(AttributeError): csaf_files.to_tuples(("not_existing",)) @@ -226,39 +238,41 @@ def test_to_tuples_exception(self, csaf_cves, csaf_files): with pytest.raises(KeyError): csaf_cves.to_tuples("wrong_key", ("cpe",)) - def test_csaf_data_assignment(self, csaf_cves, csaf_file): + def test_csaf_data_assignment(self, csaf_cves: m.CsafCves, csaf_file: m.CsafFile) -> None: """Test item assignemnt to CsafData.""" - data = model.CsafData() + data = m.CsafData() data.files[csaf_file.name] = csaf_file - data.files.update(model.CsafFiles({csaf_file.name: csaf_file})) + data.files.update(m.CsafFiles({csaf_file.name: csaf_file})) data.cves["key1"] = csaf_cves["key1"] data.cves.update(csaf_cves) - def test_cves_items_method(self): + def test_cves_items_method(self) -> None: """Test CsafCves _cves iteration with key and value pairs""" - cves_collection = model.CsafCves({ - 'key1': [model.CsafProduct(cpe='cpe123', package='kernel', status_id=4, module='module:8')], - 'key2': [model.CsafProduct(cpe='cpe456', package='nginx', status_id=4, module='web')], - }) + cves_collection = m.CsafCves( + { + "key1": m.CsafProducts([m.CsafProduct(cpe="cpe123", package="kernel", status_id=4, module="module:8")]), + "key2": m.CsafProducts([m.CsafProduct(cpe="cpe456", package="nginx", status_id=4, module="web")]), + } + ) expected_items = [ - ('key1', [model.CsafProduct(cpe='cpe123', package='kernel', status_id=4, module='module:8')]), - ('key2', [model.CsafProduct(cpe='cpe456', package='nginx', status_id=4, module='web')]), + ("key1", m.CsafProducts([m.CsafProduct(cpe="cpe123", package="kernel", status_id=4, module="module:8")])), + ("key2", m.CsafProducts([m.CsafProduct(cpe="cpe456", package="nginx", status_id=4, module="web")])), ] items_result = cves_collection.items() assert list(items_result) == expected_items - def test_keys_empty_dictionary(self): + def test_keys_empty_dictionary(self) -> None: """Test the keys method with an empty dictionary""" - csaf_cves = model.CsafCves() + csaf_cves = m.CsafCves() result = csaf_cves.keys() assert not result - def test_keys_non_empty_dictionary(self): + def test_keys_non_empty_dictionary(self) -> None: """Test the keys method with a non-empty dictionary""" key1 = "key1" key2 = "key2" - csaf_cves = model.CsafCves({key1: [], key2: []}) + csaf_cves = m.CsafCves({key1: m.CsafProducts(), key2: m.CsafProducts()}) result = csaf_cves.keys() - assert result == KeysView([key1, key2]) + assert result == {key1: "", key2: ""}.keys() From 6a129c2d030672aeaf321e2cf40d69b26b29f256 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Wed, 13 Mar 2024 17:23:17 +0100 Subject: [PATCH 08/14] fix(csaf): don't guess cve name by file name --- vmaas/reposcan/database/csaf_store.py | 11 +++++------ vmaas/reposcan/redhatcsaf/csaf_controller.py | 4 +++- vmaas/reposcan/redhatcsaf/modeling.py | 3 +++ 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/vmaas/reposcan/database/csaf_store.py b/vmaas/reposcan/database/csaf_store.py index 3746eeb2b..0f766d21b 100644 --- a/vmaas/reposcan/database/csaf_store.py +++ b/vmaas/reposcan/database/csaf_store.py @@ -1,7 +1,6 @@ """ Module containing classes for fetching/importing CSAF data from/into database. """ -import re from copy import deepcopy from typing import Any @@ -17,7 +16,6 @@ from vmaas.reposcan.mnm import CSAF_FAILED_UPDATE from vmaas.reposcan.redhatcsaf import modeling as model -CVE_PATTERN = re.compile(r"cve-\d+-\d+", re.IGNORECASE) FAILED_FILE_DELETE = "Failure while deleting CSAF file" FAILED_FILE_IMPORT = "Failed to import csaf file to DB" FAILED_PRODUCT_FETCH = "Failure while fetching CSAF products from the DB" @@ -83,10 +81,11 @@ def _save_csaf_files(self, csaf_files: model.CsafFiles) -> None: finally: cur.close() for row in rows: - # translate file name to cve, e.g. 2023/cve-2023-0030.json -> CVE-2023-0030 - match = re.search(CVE_PATTERN, row[1]) - if match: - cve = match[0].upper() + file_cves = csaf_files[row[1]].cves + if not file_cves: + self.logger.warning("File %s not associated with any CVEs", row[1]) + continue + for cve in file_cves: self.cve2file_id[cve] = row[0] def _get_product_attr_id( # type: ignore[return] diff --git a/vmaas/reposcan/redhatcsaf/csaf_controller.py b/vmaas/reposcan/redhatcsaf/csaf_controller.py index 57ed96090..4177694d7 100644 --- a/vmaas/reposcan/redhatcsaf/csaf_controller.py +++ b/vmaas/reposcan/redhatcsaf/csaf_controller.py @@ -114,8 +114,10 @@ def store(self) -> None: to_store = CsafData() for csaf_file in batch: + parsed_cves = self.parse_csaf_file(csaf_file) + csaf_file.cves = list(parsed_cves.keys()) to_store.files[csaf_file.name] = csaf_file - to_store.cves.update(self.parse_csaf_file(csaf_file)) + to_store.cves.update(parsed_cves) self.csaf_store.store(to_store) finally: diff --git a/vmaas/reposcan/redhatcsaf/modeling.py b/vmaas/reposcan/redhatcsaf/modeling.py index 98b9cc60a..b737776fd 100644 --- a/vmaas/reposcan/redhatcsaf/modeling.py +++ b/vmaas/reposcan/redhatcsaf/modeling.py @@ -37,6 +37,9 @@ class CsafFile: csv_timestamp: datetime db_timestamp: datetime | None = None id_: int = 0 + # it should be 1:1 mapping between file and cve + # but the json in file contains list of cves so add the whole list + cves: list[str] | None = None @property def out_of_date(self) -> bool: From 93b69e86e2bfc26b04005a946904348be09da84e Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Thu, 14 Mar 2024 11:10:59 +0100 Subject: [PATCH 09/14] chore(csaf): extend model tests --- .../reposcan/redhatcsaf/test/test_modeling.py | 168 +++++++++++++++--- 1 file changed, 147 insertions(+), 21 deletions(-) diff --git a/vmaas/reposcan/redhatcsaf/test/test_modeling.py b/vmaas/reposcan/redhatcsaf/test/test_modeling.py index 3cb3a9d17..58bd4680a 100644 --- a/vmaas/reposcan/redhatcsaf/test/test_modeling.py +++ b/vmaas/reposcan/redhatcsaf/test/test_modeling.py @@ -10,9 +10,10 @@ import vmaas.reposcan.redhatcsaf.modeling as m DictCollection = m.CsafCves | m.CsafFiles +CsafCollection = DictCollection | m.CsafProducts -# pylint: disable=too-many-public-methods +# pylint: disable=too-many-public-methods,protected-access class TestModels: """Test CSAF models.""" @@ -22,7 +23,7 @@ def csaf_product(self) -> m.CsafProduct: return m.CsafProduct("cpe", "package", 4, "module") @pytest.fixture - def csaf_product_list(self, csaf_product: m.CsafProduct) -> m.CsafProducts: + def csaf_products(self, csaf_product: m.CsafProduct) -> m.CsafProducts: """Returns list of CSAF products.""" return m.CsafProducts([csaf_product]) @@ -34,9 +35,9 @@ def csaf_file(self) -> m.CsafFile: return m.CsafFile("name", now1d, now) @pytest.fixture - def csaf_cves(self, csaf_product_list: m.CsafProducts) -> m.CsafCves: + def csaf_cves(self, csaf_products: m.CsafProducts) -> m.CsafCves: """Returns CSAF CVEs collection.""" - return m.CsafCves({"key1": csaf_product_list, "key2": csaf_product_list}) + return m.CsafCves({"key1": csaf_products, "key2": csaf_products}) @pytest.fixture def csaf_files(self, csaf_file: m.CsafFile) -> m.CsafFiles: @@ -49,7 +50,7 @@ def csaf_product_updated(self) -> m.CsafProduct: return m.CsafProduct("cpe_updated", "package_updated", 4, "module_updated") @pytest.fixture - def csaf_product_list_updated(self, csaf_product_updated: m.CsafProduct) -> m.CsafProducts: + def csaf_products_updated(self, csaf_product_updated: m.CsafProduct) -> m.CsafProducts: """Returns list of CSAF products.""" return m.CsafProducts([csaf_product_updated]) @@ -61,9 +62,9 @@ def csaf_file_updated(self) -> m.CsafFile: return m.CsafFile("name_updated", now1d, now) @pytest.fixture - def csaf_cves_updated(self, csaf_product_list_updated: m.CsafProducts) -> m.CsafCves: + def csaf_cves_updated(self, csaf_products_updated: m.CsafProducts) -> m.CsafCves: """Returns CSAF CVEs collection.""" - return m.CsafCves({"key2": csaf_product_list_updated}) + return m.CsafCves({"key2": csaf_products_updated}) @pytest.fixture def csaf_files_updated(self, csaf_file_updated: m.CsafFile) -> m.CsafFiles: @@ -71,7 +72,7 @@ def csaf_files_updated(self, csaf_file_updated: m.CsafFile) -> m.CsafFiles: return m.CsafFiles({"key2": csaf_file_updated}) @pytest.mark.parametrize( - "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + "collection_name, expected_arg", (("csaf_cves", "csaf_products"), ("csaf_files", "csaf_file")) ) def test_get(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test get()""" @@ -88,7 +89,7 @@ def test_get(self, collection_name: str, expected_arg: str, request: pytest.Fixt assert res == "default" @pytest.mark.parametrize( - "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + "collection_name, expected_arg", (("csaf_cves", "csaf_products"), ("csaf_files", "csaf_file")) ) def test_getitem(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test __getitem__.""" @@ -99,6 +100,12 @@ def test_getitem(self, collection_name: str, expected_arg: str, request: pytest. with pytest.raises(KeyError): collection["key99"] # pylint: disable=pointless-statement + def test_products_getitem(self, csaf_products: m.CsafProducts, csaf_product: m.CsafProduct) -> None: + """Test __getitem__ of CsafProducts.""" + assert csaf_products[0] == csaf_product + with pytest.raises(IndexError): + csaf_products[999] # pylint: disable=pointless-statement + @pytest.mark.parametrize( "collection_name, expected_arg", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) ) @@ -109,6 +116,11 @@ def test_setitem(self, collection_name: str, expected_arg: str, request: pytest. collection["key2"] = expected["key2"] assert collection["key2"] == expected["key2"] + def test_products_setitem(self, csaf_products: m.CsafProducts, csaf_product_updated: m.CsafProduct) -> None: + """Test __setitem__ of CsafProducts.""" + csaf_products[0] = csaf_product_updated + assert csaf_products[0] == csaf_product_updated + @pytest.mark.parametrize( "collection_name, expected_arg", (("csaf_cves", "csaf_cves_updated"), ("csaf_files", "csaf_files_updated")) ) @@ -119,18 +131,19 @@ def test_update(self, collection_name: str, expected_arg: str, request: pytest.F collection.update(expected) assert collection["key2"] == expected["key2"] - @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files")) + @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files", "csaf_products")) def test_iter(self, collection_name: str, request: pytest.FixtureRequest) -> None: """Test __iter__""" - collection: DictCollection = request.getfixturevalue(collection_name) + collection: CsafCollection = request.getfixturevalue(collection_name) assert isinstance(collection, Iterable) @pytest.mark.parametrize( - "collection_name, expected_arg", (("csaf_cves", "csaf_product_list"), ("csaf_files", "csaf_file")) + "collection_name, expected_arg", + (("csaf_cves", "csaf_products"), ("csaf_files", "csaf_file"), ("csaf_products", "csaf_product")), ) def test_next(self, collection_name: str, expected_arg: str, request: pytest.FixtureRequest) -> None: """Test __next__""" - collection: DictCollection = request.getfixturevalue(collection_name) + collection: CsafCollection = request.getfixturevalue(collection_name) expected = request.getfixturevalue(expected_arg) assert next(collection) == expected @@ -141,8 +154,20 @@ def test_contains(self, collection_name: str, request: pytest.FixtureRequest) -> assert "key1" in collection assert "key99" not in collection + def test_products_contains(self, csaf_products: m.CsafProducts, csaf_product: m.CsafProduct) -> None: + """Test __contains of CsafProducts.""" + assert csaf_product in csaf_products + assert m.CsafProduct("not_cpe", "not_package", 9) not in csaf_products + @pytest.mark.parametrize( - "obj, expected", (("csaf_cves", "key1"), ("csaf_files", "key1"), ("csaf_product", "cpe"), ("csaf_file", "name")) + "obj, expected", + ( + ("csaf_cves", "key1"), + ("csaf_files", "key1"), + ("csaf_product", "cpe"), + ("csaf_file", "name"), + ("csaf_products", "CsafProduct"), + ), ) def test_repr(self, obj: str, expected: str, request: pytest.FixtureRequest) -> None: """Test __repr__""" @@ -151,12 +176,12 @@ def test_repr(self, obj: str, expected: str, request: pytest.FixtureRequest) -> assert isinstance(res, str) assert expected in res - @pytest.mark.parametrize("collection_name", ("csaf_cves", "csaf_files")) - def test_len(self, collection_name: str, request: pytest.FixtureRequest) -> None: + @pytest.mark.parametrize("collection_name, expected", (("csaf_cves", 2), ("csaf_files", 2), ("csaf_products", 1))) + def test_len(self, collection_name: str, expected: int, request: pytest.FixtureRequest) -> None: """Test __len__""" - collection: DictCollection = request.getfixturevalue(collection_name) + collection: CsafCollection = request.getfixturevalue(collection_name) res = len(collection) - assert res == 2 + assert res == expected @pytest.mark.parametrize( "class_name, args", @@ -166,6 +191,7 @@ def test_len(self, collection_name: str, request: pytest.FixtureRequest) -> None ("CsafProduct", ("x", "y", 4, "z")), ("CsafCves", None), ("CsafData", None), + ("CsafProducts", None), ), ) def test_instantiate(self, class_name: str, args: tuple[str | int] | None) -> None: @@ -209,13 +235,17 @@ def test_from_table_map_and_csv(self, tmp_path: Path) -> None: @pytest.mark.parametrize( "collection_name, attr_tuple, by_key", - (("csaf_cves", ("cpe", "package", "module"), "key1"), ("csaf_files", ("name",), None)), + ( + ("csaf_cves", ("cpe", "package", "module"), "key1"), + ("csaf_products", ("cpe", "package", "module"), None), + ("csaf_files", ("name",), None), + ), ) def test_to_tuples( self, collection_name: str, attr_tuple: tuple[str], by_key: str | None, request: pytest.FixtureRequest ) -> None: """Test collection.to_tuples()""" - collection: DictCollection = request.getfixturevalue(collection_name) + collection: CsafCollection = request.getfixturevalue(collection_name) def _assert_tuples(item: tuple[Any, ...]) -> None: assert len(item) == len(attr_tuple) @@ -229,12 +259,16 @@ def _assert_tuples(item: tuple[Any, ...]) -> None: files_tuple, *_ = collection.to_tuples(attr_tuple) _assert_tuples(files_tuple) - def test_to_tuples_exception(self, csaf_cves: m.CsafCves, csaf_files: m.CsafFiles) -> None: + def test_to_tuples_exception( + self, csaf_cves: m.CsafCves, csaf_files: m.CsafFiles, csaf_products: m.CsafProducts + ) -> None: """Test exceptions from collection.to_tuples()""" with pytest.raises(AttributeError): csaf_files.to_tuples(("not_existing",)) with pytest.raises(AttributeError): csaf_cves.to_tuples("key1", ("not_existing",)) + with pytest.raises(AttributeError): + csaf_products.to_tuples(("not_existing",)) with pytest.raises(KeyError): csaf_cves.to_tuples("wrong_key", ("cpe",)) @@ -276,3 +310,95 @@ def test_keys_non_empty_dictionary(self) -> None: csaf_cves = m.CsafCves({key1: m.CsafProducts(), key2: m.CsafProducts()}) result = csaf_cves.keys() assert result == {key1: "", key2: ""}.keys() + + @pytest.fixture + def csaf_products_with_ids(self) -> m.CsafProducts: + """CsafProducts with multiple values with ids.""" + return m.CsafProducts( + [ + # with ids + m.CsafProduct("cpe1", "pkg1", 1, "module1", 1, 1, 1, None), + m.CsafProduct("cpe2", "pkg2", 2, "module2", 2, 2, 2, None), + # missing product id + m.CsafProduct("cpe3", "pkg3", 3, "module3", None, 3, 3, None), + # missing cpe_id + m.CsafProduct("cpe4", "pkg4", 4, "module4", 4, None, 4, None), + # missing package id + m.CsafProduct("cpe5", "pkg5", 5, "module5", 5, 5, None, None), + ] + ) + + def test_products_add_lookup(self, csaf_products_with_ids: m.CsafProducts) -> None: + """Test CsafProducts add_to_lookup.""" + products = csaf_products_with_ids + for product in products: + assert product not in products._lookup.values() + for product in products: + if product.cpe_id: + products.add_to_lookup(product) + assert product in products._lookup.values() + + def test_products_get(self, csaf_products_with_ids: m.CsafProducts) -> None: + """Test CsafProducts get_by_ids.""" + products = csaf_products_with_ids + # get from _products (and add to lookup) + for prod in products: + if prod.cpe_id: + assert prod not in products._lookup.values() + res = products.get_by_ids_and_module(prod.cpe_id, prod.package_name_id, prod.package_id, prod.module) + assert res == prod + # get from _lookup (added by previous get_by_ids call) + for prod in products: + if prod.cpe_id: + assert prod in products._lookup.values() + res = products.get_by_ids_and_module(prod.cpe_id, prod.package_name_id, prod.package_id, prod.module) + assert res == prod + + def test_products_append(self, csaf_products: m.CsafProducts) -> None: + """Test CsafProducts append.""" + product = m.CsafProduct("cpe_append", "pkg_append", 1, cpe_id=1) + assert product not in csaf_products + csaf_products.append(product) + assert product in csaf_products + assert product in csaf_products._products + assert product in csaf_products._lookup.values() + + def test_products_remove(self, csaf_products: m.CsafProducts) -> None: + """Test CsafProducts remove.""" + product = m.CsafProduct("cpe_remove", "pkg_remove", 1) + + def _assert_not_in_products() -> None: + assert product not in csaf_products + assert product not in csaf_products._products + assert product not in csaf_products._lookup.values() + + _assert_not_in_products() + with pytest.raises(ValueError): + csaf_products.remove(product) + + csaf_products.append(product) + assert product in csaf_products + + csaf_products.remove(product) + _assert_not_in_products() + + @pytest.mark.parametrize( + "filter_, expected", + ( + ("missing_only", [("cpe3",)]), + ("with_id", [("cpe1",), ("cpe2",), ("cpe4",), ("cpe5",)]), + ("with_cpe_id", [("cpe1",), ("cpe2",), ("cpe3",), ("cpe5",)]), + ("with_pkg_id", [("cpe1",), ("cpe2",), ("cpe3",), ("cpe4",)]), + ("with_all", [("cpe1",), ("cpe2",)]), + ), + ) + def test_products_tuples_filters( + self, filter_: str, expected: list[tuple[str]], csaf_products_with_ids: m.CsafProducts + ) -> None: + """Test filters in CsafProducts to_tuples.""" + products = csaf_products_with_ids + kwargs = {filter_: True} + if filter_ == "with_all": + kwargs = {"with_id": True, "with_cpe_id": True, "with_pkg_id": True} + tuples = products.to_tuples(("cpe",), **kwargs) + assert tuples == expected From 8850d9e86a80599854aaab88eaaeb49dbe82e6e8 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Fri, 15 Mar 2024 17:19:14 +0100 Subject: [PATCH 10/14] chore(csaf): add csaf_store tests --- pyproject.toml | 1 + vmaas/reposcan/conftest.py | 2 +- vmaas/reposcan/database/test/test_csaf.py | 191 ++++++++++++++++++++++ 3 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 vmaas/reposcan/database/test/test_csaf.py diff --git a/pyproject.toml b/pyproject.toml index 73c50889b..68b20dbf3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -62,5 +62,6 @@ warn_unreachable = true follow_imports = "silent" packages = """ vmaas.reposcan.database.csaf_store, + vmaas.reposcan.database.test.test_csaf, vmaas.reposcan.redhatcsaf """ diff --git a/vmaas/reposcan/conftest.py b/vmaas/reposcan/conftest.py index 7d383b545..16373f89d 100644 --- a/vmaas/reposcan/conftest.py +++ b/vmaas/reposcan/conftest.py @@ -86,7 +86,7 @@ def reset_db(conn, old_schema: bool = False): conn.commit() -def write_testing_data(conn): +def write_testing_data(conn: psycopg2.extensions.connection) -> None: """Write testing data to the database.""" with conn.cursor() as cursor: cursor.execute(VMAAS_DB_DATA.read_text(encoding="utf-8")) diff --git a/vmaas/reposcan/database/test/test_csaf.py b/vmaas/reposcan/database/test/test_csaf.py new file mode 100644 index 000000000..45aeb0312 --- /dev/null +++ b/vmaas/reposcan/database/test/test_csaf.py @@ -0,0 +1,191 @@ +"""Unit tests of csaf_store.py.""" +import typing as t +from datetime import datetime +from datetime import timezone + +import pytest +from psycopg2.extensions import connection +from psycopg2.extras import execute_values + +import vmaas.reposcan.redhatcsaf.modeling as m +from vmaas.reposcan.conftest import reset_db +from vmaas.reposcan.conftest import write_testing_data +from vmaas.reposcan.database.csaf_store import CsafStore + + +EXISTING_PRODUCTS = [ + m.CsafProduct("cpe1000", "pkg1000", 4, None), + m.CsafProduct("cpe1001", "pkg1001-1:1-1.noarch", 3, None), + m.CsafProduct("cpe1002", "pkg1002", 4, "module:1"), + m.CsafProduct("cpe1003", "pkg1003-1:1-1.noarch", 3, "module:1"), +] +NEW_PRODUCTS = [ + m.CsafProduct("cpe1000", "pkg1001", 4, None), + m.CsafProduct("cpe1001", "pkg1000-1:1-1.noarch", 3, None), + m.CsafProduct("cpe1002", "pkg1003", 4, "module:1"), + m.CsafProduct("cpe1003", "pkg1002-1:1-1.noarch", 3, "module:1"), + m.CsafProduct("cpe1002", "pkg1003", 4, "module:2"), + m.CsafProduct("cpe1003", "pkg1002-1:1-1.noarch", 3, "module:2"), +] + + +# pylint: disable=protected-access +class TestCsafStore: + """CsafStore tests.""" + + @pytest.fixture + def csaf_store(self, db_conn: connection) -> t.Generator[CsafStore, None, None]: # pylint: disable=unused-argument + """Fixture returning CsafStore obj and cleaning db after test.""" + store = CsafStore() + yield store + reset_db(store.conn) + + @pytest.fixture + def products(self, csaf_store: CsafStore) -> t.Generator[None, None, None]: + """Setup products in DB.""" + timestamp = datetime.now() + write_testing_data(csaf_store.conn) + cpes = ((1000, "cpe1000"), (1001, "cpe1001"), (1002, "cpe1002"), (1003, "cpe1003")) + package_names = ((1000, "pkg1000"), (1001, "pkg1001"), (1002, "pkg1002"), (1003, "pkg1003")) + packages = ( + (1000, 1000, 201, 1, timestamp), + (1001, 1001, 201, 1, timestamp), + (1002, 1002, 201, 1, timestamp), + (1003, 1003, 201, 1, timestamp), + ) + products = ( + (1000, 1000, 1000, None, None), + (1001, 1001, None, 1001, None), + (1002, 1002, 1002, None, "module:1"), + (1003, 1003, None, 1003, "module:1"), + ) + cur = csaf_store.conn.cursor() + execute_values(cur, "INSERT INTO csaf_file(id, name, updated) VALUES %s RETURNING id", ((1, "file1", timestamp),)) + execute_values(cur, "INSERT INTO cpe(id, label) VALUES %s RETURNING id", cpes) + execute_values(cur, "INSERT INTO package_name(id, name) VALUES %s RETURNING id", package_names) + execute_values( + cur, "INSERT INTO package(id, name_id, evr_id, arch_id, modified) VALUES %s RETURNING id", packages + ) + execute_values( + cur, + "INSERT INTO csaf_product(id, cpe_id, package_name_id, package_id, module_stream) VALUES %s RETURNING id", + products, + ) + csaf_store.conn.commit() + cur.close() + + yield + + reset_db(csaf_store.conn) + + def test_save_file(self, csaf_store: CsafStore) -> None: + """Test saving csaf file.""" + now = datetime.now(timezone.utc) + csaf_store._save_csaf_files(m.CsafFiles({"file": m.CsafFile("file", now, cves=["CVE-2024-1234"])})) + cur = csaf_store.conn.cursor() + cur.execute("SELECT id, name FROM csaf_file WHERE name = 'file'") + res = cur.fetchone() + assert res + id_save = res[0] + assert "CVE-2024-1234" in csaf_store.cve2file_id + assert csaf_store.cve2file_id["CVE-2024-1234"] == id_save + + # update row + update_ts = datetime.now(timezone.utc) + csaf_store._save_csaf_files(m.CsafFiles({"file": m.CsafFile("file", update_ts, cves=["CVE-2024-1234"])})) + cur.execute("SELECT id, updated FROM csaf_file WHERE name = 'file'") + res = cur.fetchone() + assert res + assert res[0] == id_save + assert res[1] == update_ts + + def test_get_product_attr_id(self, csaf_store: CsafStore) -> None: + """Test getting product attribute_id.""" + mapping = {"key": 9} + res = csaf_store._get_product_attr_id("some_attr", mapping, "key") + assert res == 9 + + with pytest.raises(KeyError): + csaf_store._get_product_attr_id("some_attr", mapping, "bad_key") + + def test_load_product_attr_ids(self, products: None) -> None: # pylint: disable=unused-argument + """Test loading product atrribute_id.""" + csaf_store = CsafStore() + products_obj = m.CsafProducts( + EXISTING_PRODUCTS + + [ + # will be skipped - missing cpe + m.CsafProduct("cpe_missing", "pkg1000", 4, None), + # will be skipped - missing package + m.CsafProduct("cpe1000", "pkg_missing", 4, None), + ] + ) + csaf_store._load_product_attr_ids(products_obj) + assert len(products_obj) == 4 + for product in products_obj: + assert product.cpe_id + assert product.package_name_id or product.package_id + + def test_update_product_ids(self, products: None) -> None: # pylint: disable=unused-argument + """Test loading product ids.""" + csaf_store = CsafStore() + products_obj = m.CsafProducts(EXISTING_PRODUCTS + NEW_PRODUCTS) + csaf_store._update_product_ids(products_obj) + for i, product in enumerate(products_obj): + if i < len(EXISTING_PRODUCTS): + assert product.id_ + else: + assert not product.id_ + + def test_insert_missing_products(self, products: None) -> None: # pylint: disable=unused-argument + """Test inserting missing product ids.""" + csaf_store = CsafStore() + products_obj = m.CsafProducts(EXISTING_PRODUCTS + NEW_PRODUCTS) + csaf_store._update_product_ids(products_obj) + csaf_store._insert_missing_products(products_obj) + ids = [] + for product in products_obj: + assert product.id_ + assert product.id_ not in ids + ids.append(product.id_) + + def assert_cve_count(self, csaf_store: CsafStore, count: int) -> None: + """Assert cve count in db""" + cur = csaf_store.conn.cursor() + cur.execute("SELECT id FROM cve WHERE UPPER(name) = %s", ("CVE-0000-0001",)) + cve_id = cur.fetchone()[0] + cur.execute("SELECT count(*) FROM csaf_cve_product WHERE cve_id = %s", (cve_id,)) + db_count = cur.fetchone()[0] + assert db_count == count + + def test_insert_cves(self, products: None) -> None: # pylint: disable=unused-argument + """Test inserting csaf_cve_product.""" + csaf_store = CsafStore() + csaf_store.cve2file_id["CVE-0000-0001"] = 1 + products_obj = m.CsafProducts(EXISTING_PRODUCTS) + csaf_store._update_product_ids(products_obj) + csaf_store._insert_cves("CVE-0000-0001", products_obj) + self.assert_cve_count(csaf_store, 4) + + def test_insert_cves_no_product_ids(self, products: None) -> None: # pylint: disable=unused-argument + """Test inserting csaf_cve_product.""" + csaf_store = CsafStore() + csaf_store.cve2file_id["CVE-0000-0001"] = 1 + products_obj = m.CsafProducts([m.CsafProduct("cpe1000", "pkg1000", 4, None)]) + csaf_store._insert_cves("CVE-0000-0001", products_obj) + self.assert_cve_count(csaf_store, 0) + + def test_remove_cves(self, products: None) -> None: # pylint: disable=unused-argument + """Test removing csaf_cve_product.""" + csaf_store = CsafStore() + csaf_store.cve2file_id["CVE-0000-0001"] = 1 + products_obj = m.CsafProducts([m.CsafProduct("cpe1000", "pkg1000", 4, None)]) + csaf_store._update_product_ids(products_obj) + csaf_store._insert_cves("CVE-0000-0001", products_obj) + self.assert_cve_count(csaf_store, 1) + + # remove old cve-product + products_obj = m.CsafProducts([m.CsafProduct("cpe1001", "pkg1001-1:1-1.noarch", 3, None)]) + csaf_store._update_product_ids(products_obj) + csaf_store._remove_cves("CVE-0000-0001", products_obj) + self.assert_cve_count(csaf_store, 0) From 161bee36746df28c6ea34fb949661686af8fb4d2 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Fri, 15 Mar 2024 17:21:51 +0100 Subject: [PATCH 11/14] chore(tests): capture logs in tests --- run_tests.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/run_tests.sh b/run_tests.sh index 9cebe0f68..5a8795753 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -24,7 +24,7 @@ for test_dir in $test_dirs; do done # Find and run tests -pytest -vvv --cov-report=xml --cov=. --color=yes --durations=1 +pytest -vvv --cov-report=xml --cov=. --color=yes --durations=1 -rP rc=$(($rc+$?)) From ce61a92bd2709765207621a97d8463f2a8ce7895 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Mon, 18 Mar 2024 14:59:27 +0100 Subject: [PATCH 12/14] fix(csaf): update file timestamp after successful cve insert/remove --- vmaas/reposcan/database/csaf_store.py | 26 ++++++-- vmaas/reposcan/database/test/test_csaf.py | 78 +++++++++++++++++++---- vmaas/reposcan/redhatcsaf/modeling.py | 7 ++ 3 files changed, 92 insertions(+), 19 deletions(-) diff --git a/vmaas/reposcan/database/csaf_store.py b/vmaas/reposcan/database/csaf_store.py index 0f766d21b..47957c59c 100644 --- a/vmaas/reposcan/database/csaf_store.py +++ b/vmaas/reposcan/database/csaf_store.py @@ -66,11 +66,11 @@ def _save_csaf_files(self, csaf_files: model.CsafFiles) -> None: rows = execute_values( cur, """ - insert into csaf_file (name, updated) values %s - on conflict (name) do update set updated = excluded.updated + insert into csaf_file (name) values %s + on conflict (name) do nothing returning id, name """, - csaf_files.to_tuples(("name", "csv_timestamp")), + csaf_files.to_tuples(("name",)), fetch=True, ) self.conn.commit() @@ -81,6 +81,7 @@ def _save_csaf_files(self, csaf_files: model.CsafFiles) -> None: finally: cur.close() for row in rows: + csaf_files[row[1]].id_ = row[0] file_cves = csaf_files[row[1]].cves if not file_cves: self.logger.warning("File %s not associated with any CVEs", row[1]) @@ -338,7 +339,21 @@ def _remove_cves(self, cve: str, products: model.CsafProducts) -> None: finally: cur.close() - def _populate_cves(self, csaf_cves: model.CsafCves) -> None: + def _update_file_timestamp(self, cve: str, files: model.CsafFiles) -> None: + cur = self.conn.cursor() + try: + file_id = self.cve2file_id[cve] + csaf_file = files.get_by_id(file_id) + if csaf_file is None: + raise CsafStoreException(f"csaf_file with id={file_id} not found") + cur.execute("UPDATE csaf_file SET updated = %s WHERE id = %s", (csaf_file.csv_timestamp, file_id)) + self.conn.commit() + except Exception as exc: + raise CsafStoreException(f"failed to update csaf_file for {cve}") from exc + finally: + cur.close() + + def _populate_cves(self, csaf_cves: model.CsafCves, files: model.CsafFiles) -> None: for cve, products in csaf_cves.items(): products_copy = deepcopy(products) # only for logging of failed cves try: @@ -346,6 +361,7 @@ def _populate_cves(self, csaf_cves: model.CsafCves) -> None: self._insert_missing_products(products) self._remove_cves(cve, products) self._insert_cves(cve, products) + self._update_file_timestamp(cve, files) except CsafStoreSkippedCVE as exc: self.logger.warning("Skipping cve: %s reason: %s", cve, exc) self.logger.debug("Skipping cve: %s products: %s reason: %s", cve, products_copy, exc) @@ -375,5 +391,5 @@ def _delete_unreferenced_products(self) -> None: def store(self, csaf_data: model.CsafData) -> None: """Store collection of CSAF files into DB.""" self._save_csaf_files(csaf_data.files) - self._populate_cves(csaf_data.cves) + self._populate_cves(csaf_data.cves, csaf_data.files) self._delete_unreferenced_products() diff --git a/vmaas/reposcan/database/test/test_csaf.py b/vmaas/reposcan/database/test/test_csaf.py index 45aeb0312..90cfb505a 100644 --- a/vmaas/reposcan/database/test/test_csaf.py +++ b/vmaas/reposcan/database/test/test_csaf.py @@ -27,6 +27,7 @@ m.CsafProduct("cpe1002", "pkg1003", 4, "module:2"), m.CsafProduct("cpe1003", "pkg1002-1:1-1.noarch", 3, "module:2"), ] +CVE = "CVE-0000-0001" # pylint: disable=protected-access @@ -60,7 +61,7 @@ def products(self, csaf_store: CsafStore) -> t.Generator[None, None, None]: (1003, 1003, None, 1003, "module:1"), ) cur = csaf_store.conn.cursor() - execute_values(cur, "INSERT INTO csaf_file(id, name, updated) VALUES %s RETURNING id", ((1, "file1", timestamp),)) + execute_values(cur, "INSERT INTO csaf_file(id, name, updated) VALUES %s RETURNING id", ((1, "file1", None),)) execute_values(cur, "INSERT INTO cpe(id, label) VALUES %s RETURNING id", cpes) execute_values(cur, "INSERT INTO package_name(id, name) VALUES %s RETURNING id", package_names) execute_values( @@ -78,10 +79,18 @@ def products(self, csaf_store: CsafStore) -> t.Generator[None, None, None]: reset_db(csaf_store.conn) + @pytest.fixture + def files_obj_for_insert(self) -> tuple[m.CsafFiles, datetime]: + """Csaf files obj for insert_cves tests.""" + now = datetime.now(timezone.utc) + files_obj = m.CsafFiles({"file": m.CsafFile("file", now, None, 1, [CVE])}) + return files_obj, now + def test_save_file(self, csaf_store: CsafStore) -> None: """Test saving csaf file.""" now = datetime.now(timezone.utc) - csaf_store._save_csaf_files(m.CsafFiles({"file": m.CsafFile("file", now, cves=["CVE-2024-1234"])})) + files = m.CsafFiles({"file": m.CsafFile("file", now, cves=["CVE-2024-1234"])}) + csaf_store._save_csaf_files(files) cur = csaf_store.conn.cursor() cur.execute("SELECT id, name FROM csaf_file WHERE name = 'file'") res = cur.fetchone() @@ -89,6 +98,8 @@ def test_save_file(self, csaf_store: CsafStore) -> None: id_save = res[0] assert "CVE-2024-1234" in csaf_store.cve2file_id assert csaf_store.cve2file_id["CVE-2024-1234"] == id_save + for file in files: + assert file.id_ # update row update_ts = datetime.now(timezone.utc) @@ -97,7 +108,7 @@ def test_save_file(self, csaf_store: CsafStore) -> None: res = cur.fetchone() assert res assert res[0] == id_save - assert res[1] == update_ts + assert res[1] is None def test_get_product_attr_id(self, csaf_store: CsafStore) -> None: """Test getting product attribute_id.""" @@ -152,40 +163,79 @@ def test_insert_missing_products(self, products: None) -> None: # pylint: disab def assert_cve_count(self, csaf_store: CsafStore, count: int) -> None: """Assert cve count in db""" cur = csaf_store.conn.cursor() - cur.execute("SELECT id FROM cve WHERE UPPER(name) = %s", ("CVE-0000-0001",)) + cur.execute("SELECT id FROM cve WHERE UPPER(name) = %s", (CVE,)) cve_id = cur.fetchone()[0] cur.execute("SELECT count(*) FROM csaf_cve_product WHERE cve_id = %s", (cve_id,)) db_count = cur.fetchone()[0] assert db_count == count - def test_insert_cves(self, products: None) -> None: # pylint: disable=unused-argument + def assert_file_timestamp(self, csaf_store: CsafStore, timestamp: datetime | None, id_: int = 1) -> None: + """Assert csaf_file timestamp after insert""" + cur = csaf_store.conn.cursor() + cur.execute("SELECT updated FROM csaf_file WHERE id = %s", (id_,)) + updated = cur.fetchone()[0] + assert updated == timestamp + + def test_insert_cves( # pylint: disable=unused-argument + self, products: None, files_obj_for_insert: tuple[m.CsafFiles, datetime] + ) -> None: """Test inserting csaf_cve_product.""" csaf_store = CsafStore() - csaf_store.cve2file_id["CVE-0000-0001"] = 1 + csaf_store.cve2file_id[CVE] = 1 + files, timestamp = files_obj_for_insert products_obj = m.CsafProducts(EXISTING_PRODUCTS) csaf_store._update_product_ids(products_obj) - csaf_store._insert_cves("CVE-0000-0001", products_obj) + csaf_store._insert_cves(CVE, products_obj) + csaf_store._update_file_timestamp(CVE, files) self.assert_cve_count(csaf_store, 4) + self.assert_file_timestamp(csaf_store, timestamp) - def test_insert_cves_no_product_ids(self, products: None) -> None: # pylint: disable=unused-argument + def test_insert_cves_no_product_ids( # pylint: disable=unused-argument + self, products: None, files_obj_for_insert: tuple[m.CsafFiles, datetime] + ) -> None: """Test inserting csaf_cve_product.""" csaf_store = CsafStore() - csaf_store.cve2file_id["CVE-0000-0001"] = 1 + csaf_store.cve2file_id[CVE] = 1 + files, timestamp = files_obj_for_insert products_obj = m.CsafProducts([m.CsafProduct("cpe1000", "pkg1000", 4, None)]) - csaf_store._insert_cves("CVE-0000-0001", products_obj) + csaf_store._insert_cves(CVE, products_obj) + csaf_store._update_file_timestamp(CVE, files) self.assert_cve_count(csaf_store, 0) + # file timestamp is updated also if CVE is skipped when no products are matched + self.assert_file_timestamp(csaf_store, timestamp) + + def test_insert_cves_unknown_cpe( # pylint: disable=unused-argument + self, products: None, files_obj_for_insert: tuple[m.CsafFiles, datetime] + ) -> None: + """Test inserting csaf_cve_product.""" + csaf_store = CsafStore() + csaf_store.cve2file_id[CVE] = 1 + files, timestamp = files_obj_for_insert + products_obj = m.CsafProducts([m.CsafProduct("cpe_unknown", "pkg1000", 4, None)]) + csaf_store._update_product_ids(products_obj) + # insert product with missing cpe + csaf_store._insert_missing_products(products_obj) + csaf_store._insert_cves(CVE, products_obj) + csaf_store._update_file_timestamp(CVE, files) + self.assert_cve_count(csaf_store, 1) + self.assert_file_timestamp(csaf_store, timestamp) - def test_remove_cves(self, products: None) -> None: # pylint: disable=unused-argument + def test_remove_cves( # pylint: disable=unused-argument + self, products: None, files_obj_for_insert: tuple[m.CsafFiles, datetime] + ) -> None: """Test removing csaf_cve_product.""" csaf_store = CsafStore() - csaf_store.cve2file_id["CVE-0000-0001"] = 1 + csaf_store.cve2file_id[CVE] = 1 + files, timestamp = files_obj_for_insert products_obj = m.CsafProducts([m.CsafProduct("cpe1000", "pkg1000", 4, None)]) csaf_store._update_product_ids(products_obj) - csaf_store._insert_cves("CVE-0000-0001", products_obj) + csaf_store._insert_cves(CVE, products_obj) self.assert_cve_count(csaf_store, 1) # remove old cve-product products_obj = m.CsafProducts([m.CsafProduct("cpe1001", "pkg1001-1:1-1.noarch", 3, None)]) csaf_store._update_product_ids(products_obj) - csaf_store._remove_cves("CVE-0000-0001", products_obj) + csaf_store._remove_cves(CVE, products_obj) + csaf_store._update_file_timestamp(CVE, files) self.assert_cve_count(csaf_store, 0) + self.assert_file_timestamp(csaf_store, timestamp) diff --git a/vmaas/reposcan/redhatcsaf/modeling.py b/vmaas/reposcan/redhatcsaf/modeling.py index b737776fd..9796be92d 100644 --- a/vmaas/reposcan/redhatcsaf/modeling.py +++ b/vmaas/reposcan/redhatcsaf/modeling.py @@ -109,6 +109,13 @@ def get(self, key: str, default: CsafFile | None = None) -> CsafFile | None: """Return the value for key if key is in the collection, else default.""" return self._files.get(key, default) + def get_by_id(self, id_: int, default: CsafFile | None = None) -> CsafFile | None: + """Return the value for id_ if id_ is in the collection, else default.""" + for csaf_file in self: + if csaf_file.id_ == id_: + return csaf_file + return default + def update(self, data: CsafFiles) -> None: """Update data in collection - same as dict.update().""" self._files.update(data._files) # pylint: disable=protected-access From 51ba2d79e4cd107ce89b02b7d0d90376c0911209 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Tue, 19 Mar 2024 12:25:20 +0100 Subject: [PATCH 13/14] fix(csaf): insert missing cpes --- vmaas/reposcan/database/cpe_store.py | 2 +- vmaas/reposcan/database/csaf_store.py | 5 +++++ vmaas/reposcan/database/test/test_csaf.py | 4 ++-- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/vmaas/reposcan/database/cpe_store.py b/vmaas/reposcan/database/cpe_store.py index 5046616cb..7ab547afd 100644 --- a/vmaas/reposcan/database/cpe_store.py +++ b/vmaas/reposcan/database/cpe_store.py @@ -42,7 +42,7 @@ def _save_lastmodified(self, lastmodified, key): cur.close() self.conn.commit() - def populate_cpes(self, cpes): # pylint: disable=too-many-branches + def populate_cpes(self, cpes: dict[str, str | None]) -> None: # pylint: disable=too-many-branches """ Insert or update CPE items. """ diff --git a/vmaas/reposcan/database/csaf_store.py b/vmaas/reposcan/database/csaf_store.py index 47957c59c..038935277 100644 --- a/vmaas/reposcan/database/csaf_store.py +++ b/vmaas/reposcan/database/csaf_store.py @@ -102,6 +102,11 @@ def _load_product_attr_ids(self, products: model.CsafProducts) -> None: for product in products: try: product.cpe_id = self._get_product_attr_id("cpe", self.cpe_store.cpe_label_to_id, value=product.cpe) + except KeyError: + self.logger.debug("Inserting missing cpe %s", product.cpe) + self.cpe_store.populate_cpes({product.cpe: None}) + product.cpe_id = self._get_product_attr_id("cpe", self.cpe_store.cpe_label_to_id, value=product.cpe) + try: if product.status_id == model.CsafProductStatus.KNOWN_AFFECTED: # product for unfixed cve, we have only package_name product.package_name_id = self._get_product_attr_id( diff --git a/vmaas/reposcan/database/test/test_csaf.py b/vmaas/reposcan/database/test/test_csaf.py index 90cfb505a..8a3dc625a 100644 --- a/vmaas/reposcan/database/test/test_csaf.py +++ b/vmaas/reposcan/database/test/test_csaf.py @@ -127,12 +127,12 @@ def test_load_product_attr_ids(self, products: None) -> None: # pylint: disable + [ # will be skipped - missing cpe m.CsafProduct("cpe_missing", "pkg1000", 4, None), - # will be skipped - missing package + # will be inserted - missing package m.CsafProduct("cpe1000", "pkg_missing", 4, None), ] ) csaf_store._load_product_attr_ids(products_obj) - assert len(products_obj) == 4 + assert len(products_obj) == len(EXISTING_PRODUCTS) + 1 # existing + missing cpe for product in products_obj: assert product.cpe_id assert product.package_name_id or product.package_id From 1f1a428ac579ec743abbdb5d826bd6559126a723 Mon Sep 17 00:00:00 2001 From: Patrik Segedy Date: Thu, 21 Mar 2024 15:26:00 +0100 Subject: [PATCH 14/14] fix(csaf): store unique products to collection --- vmaas/reposcan/redhatcsaf/csaf_controller.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/vmaas/reposcan/redhatcsaf/csaf_controller.py b/vmaas/reposcan/redhatcsaf/csaf_controller.py index 4177694d7..2c5286e86 100644 --- a/vmaas/reposcan/redhatcsaf/csaf_controller.py +++ b/vmaas/reposcan/redhatcsaf/csaf_controller.py @@ -151,7 +151,7 @@ def _parse_vulnerabilities(self, csaf: dict[str, t.Any], product_cpe: dict[str, continue cve = vulnerability["cve"].upper() - unfixed_cves[cve] = CsafProducts() + uniq_products = {} for product_status in self.cfg.csaf_product_status_list: status_id = CsafProductStatus[product_status.upper()].value for unfixed in vulnerability["product_status"].get(product_status.lower(), []): @@ -163,7 +163,8 @@ def _parse_vulnerabilities(self, csaf: dict[str, t.Any], product_cpe: dict[str, module, pkg_name = rest.split("/", 1) csaf_product = CsafProduct(product_cpe[branch_product], pkg_name, status_id, module) - unfixed_cves[cve].append(csaf_product) + uniq_products[(product_cpe[branch_product], pkg_name, module)] = csaf_product + unfixed_cves[cve] = CsafProducts(list(uniq_products.values())) return unfixed_cves