diff --git a/scripts/cve_check.py b/scripts/cve_check.py index b4e1a4c1..d0bf35e7 100755 --- a/scripts/cve_check.py +++ b/scripts/cve_check.py @@ -7,49 +7,43 @@ # # SPDX-License-Identifier: MIT # + +from lib.python.cve.plugin.eml_cve_plugin_base import EmlCvePlugin +import lib.python.cve.common_libs as cl +from lib.python.cve.nvd_lib import CveCheckMergedList, NvdCveInfoListCreator +from lib.python.cve.cve_reporter import CveReporter +from lib.python.package_info import PackageInfoHelper, PackageList +from lib.python.cve.cve_product import CveProductList +from lib.python.cve.kev_info import KevInfoList +import lib.python.cve.kev_cve as kev_cve +import lib.python.bitbake_runner as bitbake_runner + import argparse import sys import os, os.path -import sqlite3 import yaml -import logging -import debian.debian_support +from typing import Any +import pathlib +import traceback import re +import logging -logging.basicConfig(level = logging.INFO, format='%(asctime)s:%(levelname)s: %(message)s') +logging.basicConfig(level=logging.INFO, format="%(asctime)s:%(levelname)s: %(message)s") logger = logging.getLogger("emlinux-cve-check") -sys.path.append(os.path.join(os.path.dirname(__file__), 'lib/python')) -sys.path.append(os.path.join(os.path.dirname(__file__), 'lib/python/cve')) - -import nvd_cve -import debian_cve -import kernel_cve -import kev_cve -import bitbake_runner -import json - -def read_json(jsonfile): - with open(jsonfile, "r") as f: - return json.loads(f.read()) +import glob -def get_cve_products(extra_cve_product): - name = os.path.join(os.path.dirname(__file__), "../conf/cve/cve_products.yml") +import importlib.util +from concurrent.futures import ThreadPoolExecutor, as_completed - data = None - with open(name, "r") as f: - data = yaml.safe_load(f) - if extra_cve_product: - with open(extra_cve_product, "r") as f: - tmp = yaml.safe_load(f) - if tmp: - data.update(tmp) - - return data - -def get_cve_ignore(uniq_installed_pkgs, debian_codename, extra_cve_check_ignore): - name = os.path.join(os.path.dirname(__file__), "../conf/cve/cve_check_ignore.yml") +def create_ignore_list( + emlinux_layer_dir: str, + installed_packages: PackageList, + debian_codename: str, + extra_cve_check_ignore: str, +): + name = f"{emlinux_layer_dir}/conf/cve/cve_check_ignore.yml" def read_ignore_list(filename): if filename is None: @@ -72,7 +66,7 @@ def make_backward_compatibility(data): tmp = {} for pkg in data: if type(data[pkg]) is list: - tmp[pkg] = { "all": data[pkg] } + tmp[pkg] = {"all": data[pkg]} else: tmp[pkg] = data[pkg] return tmp @@ -106,9 +100,9 @@ def make_backward_compatibility(data): ignore_list[pkg] = [] all_used.append(pkg) else: - if not pkg in uniq_installed_pkgs: + if not pkg in installed_packages: continue - pkg_version = str(uniq_installed_pkgs[pkg]["upstream_version"]) + pkg_version = str(installed_packages.get_upstream_version(pkg)) if str(version) == pkg_version: ignore_list[pkg] = d[version] break @@ -125,657 +119,371 @@ def make_backward_compatibility(data): ignore_list[pkg].extend(tmp_merge_list[pkg]["all"]) return ignore_list -def update_ignored_cves_status(cves, cve_ignore_list): - if cve_ignore_list is None: - return cves - - for pkg in cve_ignore_list: - if pkg in cves: - pkg_cves = cves[pkg] - for cve in cve_ignore_list[pkg]: - if cve in pkg_cves: - cves[pkg][cve]["CVE STATUS"] = "Ignored" - return cves +def create_cve_check_merged_list( + src_pkg_names: list[str], check_results: Any +) -> CveCheckMergedList: + cve_check_merged_list = CveCheckMergedList() -def find_debian_pkg_cves(debian_cve_json, pkgs): - cvedata = read_json(debian_cve_json) - debian_pkg_cve_data = {} - cve_not_found_pkgs = [] - - # collect cves for each installed packages(based on source package name) - for name in pkgs: - if name in cvedata: - debian_pkg_cve_data[name] = cvedata[name] - else: - cve_not_found_pkgs.append(name) + cve_ids = create_cve_id_list_by_src_pkg_name_from_check_result( + src_pkg_names, check_results + ) + for src_pkg_name in src_pkg_names: + for cveid in cve_ids: + for cr in check_results: + vulns = cr[src_pkg_name] + if vulns is None: + # Plugin doesn't have CVE information for the src_pkg_name + continue + if cveid in vulns: + cve_check_merged_list.add_data( + src_pkg_name, cveid, vulns[cveid], cr.priority + ) + else: + # Plugin doesn't have CVE information for the CVE + pass + # logger.debug(f"{cveid} {src_pkg_name} is not found") + return cve_check_merged_list + + +def create_cve_id_list_by_src_pkg_name_from_check_result( + src_pkg_names: list[str], check_results: Any +) -> list[str]: + tmp_cve_ids = [] + for src_pkg_name in src_pkg_names: + for cr in check_results: + ci = cr.cve_ids_by_src_pkg(src_pkg_name) + if ci: + tmp_cve_ids.extend(ci) + return list(dict.fromkeys(tmp_cve_ids)) + + +# make a source package name list which content has CVE information +def create_src_package_name_list_from_check_result(check_results: Any) -> list[str]: + tmp_list = [] + for result in check_results: + tmp_list.extend(result.src_pkg_names()) + return list(dict.fromkeys(tmp_list)) + + +def read_recipe_source_info(deploy_dir: str) -> Any: + filepath = deploy_dir + "/all-source-info.json" + return cl.read_json(filepath) + + +def load_plugin(plugin_file: str) -> EmlCvePlugin: + path = pathlib.Path(plugin_file).resolve() + if not path.exists(): + logger.error(f"Failed to find plugin {plugin_file}") + exit(1) - return debian_pkg_cve_data, cve_not_found_pkgs + spec = importlib.util.spec_from_file_location(path.stem, str(path)) + if not spec or not spec.loader: + logger.error(f"Fail to load spec from {plugin_file}") + exit(1) -def create_unique_package(installed): - ret = {} - for pkg in installed: - src = pkg['source'] + mod = importlib.util.module_from_spec(spec) + sys.modules[path.stem] = mod + try: + spec.loader.exec_module(mod) + except Exception as e: + logger.error(e) + traceback.print_exc() + exit(1) - if src not in ret: - ret[src] = pkg - ret[src]["bin_pkgs"] = [pkg["package"]] - else: - ret[src]["bin_pkgs"].append(pkg["package"]) - - return ret - -def fill_cve_info(cves, cve_products, db_file): - conn = sqlite3.connect(db_file) - for pkgname in cves: - for cveid in cves[pkgname]: - c = conn.cursor() - sql = f"SELECT VULNSTATUS, SUMMARY, SCOREV2, SCOREV3, VECTOR, VECTORSTRING FROM NVD WHERE ID=\"{cveid}\"" - cursor = c.execute(sql) - data = cursor.fetchone() - c.close() - - # CVE id such as TEMP-0290435-0B57B5 not in NVD database. - if data: - if data[0] == "Rejected": - cves[pkgname][cveid]["CVE STATUS"] = "Rejected" - - cves[pkgname][cveid]["CVE SUMMARY"] = data[1] - cves[pkgname][cveid]["CVSS v2 BASE SCORE"] = data[2] - cves[pkgname][cveid]["CVSS v3 BASE SCORE"] = data[3] - cves[pkgname][cveid]["VECTOR"] = data[4] - cves[pkgname][cveid]["VECTOR STRING"] = data[5] - cves[pkgname][cveid]["KEV"] = "Not Found" - cves[pkgname][cveid]["MORE INFORMATION"] = f"https://nvd.nist.gov/vuln/detail/{cveid}" - else: - cves[pkgname][cveid]["CVE SUMMARY"] = "" - cves[pkgname][cveid]["CVSS v2 BASE SCORE"] = "0.0" - cves[pkgname][cveid]["CVSS v3 BASE SCORE"] = "0.0" - cves[pkgname][cveid]["VECTOR"] = "UNKNOWN" - cves[pkgname][cveid]["VECTOR STRING"] = "UNKNOWN" - cves[pkgname][cveid]["KEV"] = "Not Found" - cves[pkgname][cveid]["MORE INFORMATION"] = f"https://security-tracker.debian.org/tracker/{cveid}" - - - conn.close() - - return cves - -def status_name(fixed): - if fixed: - return "Patched" - return "Unpatched" - -def create_debian_pkg_cve_info(pkgname, bin_pkgname, installed_version, debian_cveinfo, nvd_cveinfo, codename): - ret = {} - - cve_ids = set(list(debian_cveinfo.keys()) + list(nvd_cveinfo.keys())) - for cveid in cve_ids: - fixed = False - if cveid in debian_cveinfo: - if not codename in debian_cveinfo[cveid]["releases"]: - # Package may be installed from other debian version. - logger.debug(f"pkg:{pkgname}: {cveid} for {codename} is not found") - return None - - dc = debian_cveinfo[cveid]["releases"][codename] - if dc["status"] == "resolved": - for repo in dc["repositories"]: - fixed_ver = debian.debian_support.Version(dc["repositories"][repo]) - if installed_version >= fixed_ver: - fixed = True - else: - nc = nvd_cveinfo[cveid] - fixed = nc["FIXED"] - - ret[cveid] = { - "CVE": cveid, - "PACKAGE NAME": pkgname, - "BINARY PACKAGE NAME": bin_pkgname, - "VERSION": str(installed_version), - "CVE STATUS": status_name(fixed), - } - - return ret - -def create_cve_info_by_nvd(pkgname, bin_pkgname, installed_version, nvd_cveinfo): - cve_ids = set(list(nvd_cveinfo["CVE"].keys())) - cves = nvd_cveinfo["CVE"] - ret = {} - - for cveid in cve_ids: - ret[cveid] = { - "CVE": cveid, - "PACKAGE NAME": pkgname, - "BINARY PACKAGE NAME": bin_pkgname, - "VERSION": str(installed_version), - "CVE STATUS": status_name(cves[cveid]["FIXED"]), - } - - return ret - -def merge_cve_data(uniq_installed_pkgs, installed_pkgs_cves_by_debian_data, cve_not_in_debian, installed_pkgs_cves_by_nvd_data, codename): - - cveinfo = {} - for pkgname in installed_pkgs_cves_by_debian_data: - installed_version = uniq_installed_pkgs[pkgname]["version"] - debian_cveinfo = installed_pkgs_cves_by_debian_data[pkgname] - nvd_cveinfo = None - if pkgname in installed_pkgs_cves_by_nvd_data: - if "CVE" in installed_pkgs_cves_by_nvd_data: - nvd_cveinfo = installed_pkgs_cves_by_nvd_data[pkgname] - else: - nvd_cveinfo = {} + for obj in vars(mod).values(): + if ( + isinstance(obj, type) + and issubclass(obj, EmlCvePlugin) + and obj is not EmlCvePlugin + ): + return obj - bin_pkgname = uniq_installed_pkgs[pkgname]["bin_pkgs"] + logger.info(f"Plugin is not found in {plugin_file}") - tmpret = create_debian_pkg_cve_info(pkgname, bin_pkgname, installed_version, debian_cveinfo, nvd_cveinfo, codename) - if not tmpret is None: - cveinfo[pkgname] = tmpret - else: - cve_not_in_debian.append(pkgname) + return None - for pkgname in cve_not_in_debian: - installed_version = uniq_installed_pkgs[pkgname]["version"] - nvd_cveinfo = None - if pkgname in installed_pkgs_cves_by_nvd_data: - if installed_pkgs_cves_by_nvd_data[pkgname] is None: - installed_pkgs_cves_by_nvd_data[pkgname] = {} - if "CVE" in installed_pkgs_cves_by_nvd_data[pkgname]: - nvd_cveinfo = installed_pkgs_cves_by_nvd_data[pkgname] - else: - nvd_cveinfo = {"CVE": {}} - - bin_pkgname = uniq_installed_pkgs[pkgname]["bin_pkgs"] - cveinfo[pkgname] = create_cve_info_by_nvd(pkgname, bin_pkgname, installed_version, nvd_cveinfo) - - return cveinfo - -def get_cves_by_package_from_nvd(conn, vendor, product): - c = conn.cursor() - - """ - NVD's CVE data has several data for a CVE. - e.g. - sqlite> select count(*) from products where id="CVE-2016-6321"; - 42 - sqlite> select * from products where id="CVE-2016-6321"; - ... - CVE-2016-6321|gnu|tar|1.25|=|| - CVE-2016-6321|gnu|tar|1.26|=|| - CVE-2016-6321|gnu|tar|1.27|=|| - CVE-2016-6321|gnu|tar|1.27.1|=|| - CVE-2016-6321|gnu|tar|1.28|=|| - CVE-2016-6321|gnu|tar|1.29|=|| - ... - - So, get unique CVE IDs then get CVE data by each ID. - """ - - if vendor: - sql = f"SELECT DISTINCT(ID) FROM PRODUCTS WHERE VENDOR=\"{vendor}\" AND PRODUCT=\"{product}\"" - else: - sql = f"SELECT DISTINCT(ID) FROM PRODUCTS WHERE PRODUCT=\"{product}\"" - - cursor = c.execute(sql) - - # Get unique CVE ID - cves = [] - if cursor: - for cve in cursor: - cves.append(cve[0]) - - c.close() - - return cves - -def is_fixed_in_upstream_version(upstream_version, debian_upstream_version, operator): - - if upstream_version == "": - # Can't determine affected or not. So return True to it may be affected - return False - - uver = debian_cve.parse_version(upstream_version) - dver = debian_upstream_version - - if operator == "=": - return uver == dver - elif operator == "<": - return uver <= dver - elif operator == "<=": - return uver <= dver - elif operator == ">": - return dver >= uver - elif operator == ">=": - return dver >= uver - return False # unknown operator - - -def is_affected_upstream_version(upstream_version, debian_upstream_version, operator): - - if upstream_version == "": - # Can't determine affected or not. So return True to it may be affected - return True - - uver = debian_cve.parse_version(upstream_version) - dver = debian_upstream_version - - if operator == "=": - return uver == dver - elif operator == "<": - return dver <= uver - elif operator == "<=": - return dver <= uver - elif operator == ">": - return dver >= uver - elif operator == ">=": - return dver >= uver - - return True # unknown operator - -def check_affected_upstream_version(conn, cveid, pkg, vendor, product): - """ - Check debian package's upstream version and start/end version in NVD database. - if debian's upstream version is less than start version or greator than or equal - debian's package version, this CVE may not be affected. - However, debian package may backport vulnerable feature from upstream so we can't - completely determine this CVE is affected or not. - """ - - cves = [] - c = conn.cursor() - - if vendor: - sql = f"SELECT * FROM PRODUCTS WHERE ID=\"{cveid}\" AND VENDOR=\"{vendor}\" AND PRODUCT=\"{product}\"" - else: - sql = f"SELECT * FROM PRODUCTS WHERE ID=\"{cveid}\" AND PRODUCT=\"{product}\"" - - cursor = c.execute(sql) - if cursor: - for cve in cursor: - d = { - "CVE": cve[0], - "VERSION_START": cve[3], - "VERSION_START_OPERAND": cve[4], - "VERSION_END": cve[5], - "VERSION_END_OPERAND": cve[6], - } - cves.append(d) - c.close() - - affected = False - fixed = False - - for cve in cves: - start_version = cve["VERSION_START"] - end_version = cve["VERSION_END"] - - if not start_version == "": - if not affected: - op = cve["VERSION_START_OPERAND"] - affected = is_affected_upstream_version(start_version, pkg["upstream_version"], op) - - if not end_version == "": - if not fixed: - op = cve["VERSION_END_OPERAND"] - fixed = is_fixed_in_upstream_version(end_version, pkg["upstream_version"], op) - - # if start version's operator uses "=" and end vesion is not set in nvd database, - # it may be able to set not affected(e.g. CVE-2002-0059. This CVE is not in debian security tracker) - if not affected and not fixed: - fixed = True - - return affected, fixed - -def pkg_cve_fixed_check_by_nvd_data(uniq_installed_pkgs, installed_debian_pkgs_cves, db_file, cve_products): - conn = sqlite3.connect(db_file) - - cve_analyzed = {} - - for name in uniq_installed_pkgs: - cve_analyzed[name] = None - - pkg = uniq_installed_pkgs[name] - vendor = None - product = name - if name in cve_products: - vendor = cve_products[name]["vendor"] - product = cve_products[name]["product"] - - cve_ids = get_cves_by_package_from_nvd(conn, vendor, product) +def load_plugins(plugin_files: list[str]): + plugins = [] - for cveid in cve_ids: - affected, fixed = check_affected_upstream_version(conn, cveid, pkg, vendor, product) - tmp = { - "DEBIAN_SRC_PKG_NAME": name, - "VENDOR": vendor, - "PRODICT": product, - "CVE": cveid, - "AFFECTED": affected, - "FIXED": fixed, - } - - if cve_analyzed[name] is None: - cve_analyzed[name] = { "CVE": {}, } - - cve_analyzed[name]["CVE"][cveid] = tmp - - conn.close() - - return cve_analyzed - -def recheck_kernel_cve(db_file, kernel_pkg_name, kernel_src_name, version, cveid): - conn = sqlite3.connect(db_file) - cve = {} - sql = f"SELECT VULNSTATUS, SUMMARY, SCOREV2, SCOREV3, VECTOR, VECTORSTRING FROM NVD WHERE ID=\"{cveid}\"" - - c = conn.cursor() - cursor = c.execute(sql) - data = cursor.fetchone() - c.close() - conn.close() - - cve["BINARY PACKAGE NAME"] = kernel_pkg_name, - cve["PACKAGE NAME"] = kernel_src_name - cve["VERSION"] = version - cve["CVE"] = cveid - - if data is None: - # No CVE data but it may be reserved. - cve["CVE STATUS"] = status_name(False) - cve["CVE SUMMARY"] = "" - cve["CVSS v2 BASE SCORE"] = "0.0" - cve["CVSS v3 BASE SCORE"] = "0.0" - cve["VECTOR"] = "UNKNOWN" - cve["VECTOR STRING"] = "UNKNOWN" - else: - fixed = False - if data[0] == "Rejected": - cve["CVE STATUS"] = "Rejected" - else: - cve["CVE STATUS"] = status_name(fixed) - cve["CVE SUMMARY"] = data[1] - cve["CVSS v2 BASE SCORE"] = data[2] - cve["CVSS v3 BASE SCORE"] = data[3] - cve["VECTOR"] = data[4] - cve["VECTOR STRING"] = data[5] - - cve["KEV"] = "None" - cve["MORE INFORMATION"] = f"https://nvd.nist.gov/vuln/detail/{cveid}" - - return cve - -def check_kernel_cves_by_cip_kernel_sec(uniq_installed_pkgs, cves, kernel_src_dir, cip_kernel_sec_dir, db_file): - if "linux-cip" in uniq_installed_pkgs: - kernel_name = "linux-cip" - elif "linux-cip-rt" in uniq_installed_pkgs: - kernel_name = "linux-cip-rt" - else: - logger.debug("kernel name is not linux-cip or linux-cip-rt") - logger.debug("Skip kernel CVE check by cip-kernel-sec") - return cves - - kernel_cves = cves[kernel_name] - pkgname = uniq_installed_pkgs[kernel_name]["package"] - pkg_full_version = str(uniq_installed_pkgs[kernel_name]["version"]) - kver = pkg_full_version.split("+")[0] - - cip_kernel_sec_result = kernel_cve.run_cip_kernel_sec(kernel_src_dir, kver, cip_kernel_sec_dir) - - for patched_status in cip_kernel_sec_result: - cip_kernel_sec_cves = cip_kernel_sec_result[patched_status] - for cve in cip_kernel_sec_cves: - if not cve in kernel_cves: - cveinfo = recheck_kernel_cve(db_file, pkgname, kernel_name, pkg_full_version, cve) - kernel_cves[cve] = cveinfo - - # Replace cve status by cip-kernel-sec result - if kernel_cves[cve]["CVE STATUS"] == "Rejected": - pass - elif kernel_cves[cve]["CVE STATUS"] == patched_status: - pass - else: - kernel_cves[cve]["CVE STATUS"] = patched_status + for plugin_file in plugin_files: + logger.debug(f"loading {plugin_file}") + obj = load_plugin(plugin_file) + if obj: + plugins.append(obj) - return cves + return plugins -def create_kev_data(kev_list): - kev_json = read_json(kev_list) - kev_data = {} - for kev in kev_json["vulnerabilities"]: - cve_id = kev["cveID"] - kev_data[cve_id] = kev +# Plugin file name convention: +# 1. Plugin file must be located in scripts/lib/python/cve/plugin directory +# 2. Plugin file name must be start with eml_cve_ then ends with _plugin.py +# e.g. eml_cve_myplugin_plugin.py +def find_plugins(disable_plugins: list[str]) -> list[str]: + layer_dirs = bitbake_runner.find_layers() + plugins = [] - return kev_data + plugin_dir = "/scripts/lib/python/cve/plugin/" + for ld in layer_dirs: + d = ld + plugin_dir + pattern = f"{d}/eml_cve_*_plugin.py" + for plugin in glob.glob(pattern): + p = os.path.splitext(os.path.basename(plugin))[0] + if not p in disable_plugins: + plugins.append(plugin) + else: + logger.info(f"Plugin '{p}' is disabled") -def add_kev_info(cves, kev_list): - kev_data = create_kev_data(kev_list) + return plugins - kev_cve_ids = kev_data.keys() - for pkg in cves: - pkg_cve_ids = cves[pkg].keys() - common_cves = list(set(pkg_cve_ids) & set(kev_cve_ids)) - if len(common_cves) > 0: - for cveid in common_cves: - pkg_info = cves[pkg] - pkg_info[cveid]["KEV"] = "Found" - pkg_info[cveid]["KNOWN RANSOMWARE CAMPAIGN USE"] = kev_data[cveid]["knownRansomwareCampaignUse"] +def cve_check_worker(plugin: EmlCvePlugin, args: Any): + logger.debug(f"run {plugin.plugin_name}") - return cves + if not args.skip_update: + ret = plugin.update_database() + if not ret: + raise Exception(f"{plugin.plugin_name}: Failed to update datebase") -def create_cves_info(db_file, debian_cve_list, uniq_installed_pkgs, installed_pkgs, codename, cve_products, kernel_src_dir, cip_kernel_sec_dir, kev_list): - logger.info("Checking CVEs ...") + if args.update_cve_databese_only: + return {} - installed_pkgs_cves_by_debian_data, cve_not_in_debian = find_debian_pkg_cves(debian_cve_list, uniq_installed_pkgs) - installed_pkgs_cves_by_nvd_data = pkg_cve_fixed_check_by_nvd_data(uniq_installed_pkgs, installed_pkgs_cves_by_debian_data, db_file, cve_products) + return plugin.run_check() - cves = merge_cve_data(uniq_installed_pkgs, installed_pkgs_cves_by_debian_data, cve_not_in_debian, installed_pkgs_cves_by_nvd_data, codename) - cves = fill_cve_info(cves, cve_products, db_file) +def fetch_kev_data(cve_data_dir: str) -> KevInfoList: + try: + kev_json = kev_cve.fetch_kev_data(cve_data_dir) + return KevInfoList(cl.read_json(kev_json)) + except: + return KevInfoList({}) - cves = check_kernel_cves_by_cip_kernel_sec(uniq_installed_pkgs, cves, kernel_src_dir, cip_kernel_sec_dir, db_file) - cves = add_kev_info(cves, kev_list) - return cves +def create_disable_plugins_list(user_given_plugins: str) -> list[str]: + if not user_given_plugins: + return [] -def write_text(cves, output_dir, uniq_installed_pkgs): - filenames = [] + disable_plugins = [] - for pkgname in cves: - if len(cves[pkgname]) == 0: - continue + do_not_disable = ["eml_cve_nvd_plugin"] + disable_plugins_tmp = [p.strip() for p in user_given_plugins.split(",")] + for p in disable_plugins_tmp: + if p not in do_not_disable: + disable_plugins.append(p) + else: + logger.warning(f"Plugin '{p}' cannot be disabeld") - filename = f"{output_dir}/{pkgname}" - filenames.append(filename) + return disable_plugins - with open(filename, "w") as f: - for cve in sorted(cves[pkgname]): - info = cves[pkgname][cve] - f.write(f"PACKAGE NAME: {info['PACKAGE NAME']}\n") - f.write(f"BINARY PACKAGE NAME: {' '.join(info['BINARY PACKAGE NAME'])}\n") - f.write(f"VERSION: {info['VERSION']}\n") - f.write(f"CVE: {info['CVE']}\n") - f.write(f"CVE STATUS: {info['CVE STATUS']}\n") - f.write(f"CVE SUMMARY: {info['CVE SUMMARY']}\n") - f.write(f"CVSS v2 BASE SCORE: {info['CVSS v2 BASE SCORE']}\n") - f.write(f"CVSS v3 BASE SCORE: {info['CVSS v3 BASE SCORE']}\n") - f.write(f"VECTOR: {info['VECTOR']}\n") - f.write(f"VECTOR STRING: {info['VECTOR STRING']}\n") - f.write(f"KEV: {info['KEV']}\n") - if info["KEV"] == "Found": - f.write(f"KNOWN RANSOMWARE CAMPAIGN USE: {info['KNOWN RANSOMWARE CAMPAIGN USE']}\n") - f.write(f"MORE INFORMATION: {info['MORE INFORMATION']}\n") - f.write("\n") - return filenames +def main(args: dict): + if args.verbose_output: + logger.setLevel(logging.DEBUG) -def write_json(cves, output_dir, uniq_installed_pkgs, cve_products): - filenames = [] + bitbakeinfo = bitbake_runner.get_bitbake_information(args.image_name) - for pkgname in cves: - filename = f"{output_dir}/{pkgname}_cve.json" - filenames.append(filename) + disable_plugins = create_disable_plugins_list(args.disable_plugins) - info = cves[pkgname] - pkginfo = uniq_installed_pkgs[pkgname] + dpkg_status_file = ( + bitbakeinfo["dpkg_status"] + if not args.dpkg_status_file + else args.dpkg_status_file + ) - product = pkgname - if pkgname in cve_products: - product = cve_products[pkgname]["product"] + debian_codename = args.debian_codename + if not debian_codename: + debian_codename = bitbakeinfo["image_distro"].split("-")[1] - cvesInRecord = "Yes" - if len(info) == 0: - cvesInRecord = "No" - issues = [] - else: - issues = [] - for cve in sorted(cves[pkgname]): - issues.append(cves[pkgname][cve]) - - data = { - "version": "1", - "package": [ - { - "name": pkginfo["source"], - "binary package name": pkginfo["bin_pkgs"], - "version": str(pkginfo["version"]), - "products": [ - { - "product": product, - "cvesInRecord": cvesInRecord, - }, - ], - "issue": issues, - }, - ], - } - - with open(filename, "w") as f: - json.dump(data, f, indent=4, sort_keys=False) - - return sorted(filenames) - -def write_all_in_one_text(output_dir, image_name, text_filenames): - output_file = f"{output_dir}/{image_name}_cve" - with open(output_file, "w") as out: - for filename in text_filenames: - with open(filename, "r") as f: - out.write(f.read()) - out.write("") - -def write_all_in_one_json(output_dir, image_name, json_filenames): - all_in_one_data = { - "version": "1", - "package": [], - } - - for filename in json_filenames: - with open(filename, "r") as f: - data = json.load(f) - all_in_one_data["package"].extend(data["package"]) - - output_file = f"{output_dir}/{image_name}_cve.json" - with open(output_file, "w") as f: - json.dump(all_in_one_data, f, indent=4, sort_keys=False) - -def create_directory(target): - if not os.path.exists(target): - os.makedirs(target) - -def main(args): - if args.verbose_output: - logger.setLevel(logging.DEBUG) + installed_packages = None + cve_product_list = None - bitbakeinfo = bitbake_runner.get_bitbake_information(args.image_name) - - cve_data_dl_dir = f"{bitbakeinfo['dl_dir']}/CVE" - create_directory(cve_data_dl_dir) - - predownload_url = None - if args.cve_db_predownload: - predownload_url = bitbakeinfo["cve_db_predownload"] - if predownload_url is None: - logger.error("CVE_DB_PREDOWNLOAD_URL variable should be defined in conf/local.conf") + if args.update_cve_databese_only: + logger.info("Run on database update only mode.") + else: + if not os.path.exists(dpkg_status_file): + logger.error(f"File {dpkg_status_file} is not found.") exit(1) - update_result, db_file = nvd_cve.update_nvd_db(cve_data_dl_dir, args.nvd_api_key, predownload_url) - if not update_result: - logger.critical("Faied to fetch CVE database from NVD") - exit(1) + # Read dpkg file to get installed package information + installed_packages = PackageInfoHelper.parse_dpkg_status_file( + dpkg_status_file, + debian_codename, + target_source_package=args.target_source_package, + ) + + # Check recipe's source code provenance + recipe_source_info = read_recipe_source_info(bitbakeinfo["deploy_image_dir"]) + installed_packages.merge_recipe_source_info(recipe_source_info) + + # Read cve product list + cve_product_list = CveProductList() + cve_product_list.create_product_list( + installed_packages, bitbakeinfo["emlinux_layer_dir"], args.extra_cve_product + ) + + # Read ignore list + ignore_list = create_ignore_list( + bitbakeinfo["emlinux_layer_dir"], + installed_packages, + debian_codename, + args.extra_cve_check_ignore, + ) + + cve_data_dir = f"{bitbakeinfo['dl_dir']}/CVE" + cl.create_directory(cve_data_dir) + + # Find and load plugins + plugin_files = find_plugins(disable_plugins) + plugin_objs = load_plugins(plugin_files) + + # Create plugin instance + plugins = [] + for obj in plugin_objs: + o = obj(cve_data_dir, args, bitbakeinfo, installed_packages, cve_product_list) + plugins.append(o) + + check_results = [] + max_workers = min(args.threads, len(plugins)) + + # Run all plugins + with ThreadPoolExecutor(max_workers=max_workers) as ex: + futures = [ex.submit(cve_check_worker, p, args) for p in plugins] + for f in as_completed(futures): + try: + check_results.append(f.result()) + except Exception as e: + logger.error(f"error: {e}") + traceback.print_exc() + exit(1) if args.update_cve_databese_only: - logger.info("Finish CVE database update.") + logger.info("Updating database finished.") exit(0) - debian_cve_list = debian_cve.fetch_cve_data(cve_data_dl_dir) - if debian_cve_list is None: - logger.critical("Failed to fetch CVE data from Debian") - exit(1) + # Sort CVE data by plugin priority + check_results = sorted(check_results, key=lambda d: d.priority) - kev_list = kev_cve.fetch_kev_data(cve_data_dl_dir) - if kev_list is None: - logger.critical("Failed to fetch KEV data from CISA") - exit(1) + # Merge CVE results + src_pkg_names = create_src_package_name_list_from_check_result(check_results) - cip_kernel_sec_dir = kernel_cve.fetch_cip_kernel_sec(cve_data_dl_dir) - if cip_kernel_sec_dir is None: - logger.critical("Failed to fetch kernel-cip-sec") - exit(1) + cve_check_merged_list = create_cve_check_merged_list(src_pkg_names, check_results) - installed_pkgs = debian_cve.parse_dpkg_status(bitbakeinfo["dpkg_status"]) - uniq_installed_pkgs = create_unique_package(installed_pkgs) - cve_products = get_cve_products(args.extra_cve_product) - cve_ignore_list = get_cve_ignore(uniq_installed_pkgs, args.debian_codename, args.extra_cve_check_ignore) + cve_check_merged_list.apply_ignore_list_info(ignore_list) - linux_kernel_src_dir = os.path.abspath(bitbakeinfo["kernel_srcdir"]) - cves = create_cves_info(db_file, debian_cve_list, uniq_installed_pkgs, installed_pkgs, args.debian_codename, cve_products, linux_kernel_src_dir, cip_kernel_sec_dir, kev_list) + # Load KEV data + kev_info_list = fetch_kev_data(cve_data_dir) - cves = update_ignored_cves_status(cves, cve_ignore_list) + # Create CVE report data + creator = NvdCveInfoListCreator(cve_data_dir, installed_packages, kev_info_list) + creator.create_cve_info_list(cve_check_merged_list) - output_base_dir = f"{bitbakeinfo['deploy_dir']}/cve/{bitbakeinfo['image_full_name']}" + cve_info_list = creator.get_nvd_info_list() - create_directory(output_base_dir) + # Write CVE report + output_base_dir = ( + f"{bitbakeinfo['deploy_dir']}/cve/{bitbakeinfo['image_full_name']}" + ) - formats = args.output_format.split(",") - for fmt in formats: - if fmt == "text": - text_output_dir = f"{output_base_dir}/text" - create_directory(text_output_dir) - text_filenames = write_text(cves, text_output_dir, uniq_installed_pkgs) - write_all_in_one_text(output_base_dir, bitbakeinfo["image_full_name"], text_filenames) - elif fmt == "json": - json_output_dir = f"{output_base_dir}/json" - create_directory(json_output_dir) - json_filenames = write_json(cves, json_output_dir, uniq_installed_pkgs, cve_products) - write_all_in_one_json(output_base_dir, bitbakeinfo["image_full_name"], json_filenames) + reporter = CveReporter(output_base_dir, bitbakeinfo["image_full_name"]) + reporter.write_report(args.output_format, cve_info_list, installed_packages) - logger.info(f"CVE check finished. CVE check results are stored in {output_base_dir}") def parse_options(): parser = argparse.ArgumentParser() - - parser.add_argument("--nvd-api-key", dest="nvd_api_key", help="API key for NVD API", - metavar="NVDAPIKEY") - parser.add_argument("--debian-codename", dest="debian_codename", help="debian codename(Debian 12 is bookworm)", - default="bookworm", metavar="DEBIANCODENAME") - parser.add_argument("--output-format", dest="output_format", help="output format. available formats are text, json. formats can be comma separated string(e.g. text,json)", - default="text", metavar="OUTPUTFORMAT") - parser.add_argument("--cve-product", dest="extra_cve_product", help="User defined cve-product file", - metavar="CVEPRODUCT") - parser.add_argument("--cve-ignore", dest="extra_cve_check_ignore", help="User defined cve-check-ignore file", - metavar="CVEPRODUCT") - parser.add_argument("--image-name", dest="image_name", help="EMLinux image name", - metavar="IMAGENAME", required=True) - parser.add_argument("--cve-db-predownload", dest="cve_db_predownload", action="store_true", help="Enable CVE database predownload.URL should be defined by CVE_DB_PREDOWNLOAD_URL in conf/local.conf.") - parser.add_argument("--update-cve-databese-only", dest="update_cve_databese_only", default=False, action="store_true", - help="Do not run cve check. Update CVE database only.") - parser.add_argument("--verbose", dest="verbose_output", help="Enable verbose output", - default=False, action="store_true") - + plugin_opts = parser.add_argument_group("arguments for plugins") + cve_check_opts = parser.add_argument_group("arguments for cve check") + + # misc options + parser.add_argument( + "--verbose", + dest="verbose_output", + help="Enable verbose output", + default=False, + action="store_true", + ) + + # CVE check core options + cve_check_opts.add_argument( + "--debian-codename", + dest="debian_codename", + help="debian codename(bookworm, trixie, and etc)", + metavar="DEBIANCODENAME", + ) + cve_check_opts.add_argument( + "--output-format", + dest="output_format", + help="output format. available formats are text, json. formats can be comma separated string(e.g. text,json)", + default="text", + metavar="OUTPUTFORMAT", + ) + cve_check_opts.add_argument( + "--cve-product", + dest="extra_cve_product", + help="User defined cve-product file", + metavar="CVEPRODUCT", + ) + cve_check_opts.add_argument( + "--cve-ignore", + dest="extra_cve_check_ignore", + help="User defined cve-check-ignore file", + metavar="CVEIGNORE", + ) + cve_check_opts.add_argument( + "--image-name", + dest="image_name", + help="EMLinux image name(e.g. emlinux-image-base, emlinux-image-weston)", + metavar="IMAGENAME", + required=True, + ) + cve_check_opts.add_argument( + "--target-source-package", + dest="target_source_package", + help="Only check given debian source package(e.g. bash, util-linux", + metavar="DEBIAN SOURCE PACKAGE NAME", + ) + cve_check_opts.add_argument( + "--dpkg-status-file", + dest="dpkg_status_file", + help="Use specific dpkg_status file instead of default", + metavar="DPKG STATUS FILE", + ) + cve_check_opts.add_argument( + "--threads", default=1, help="Number of thread for cve check" + ) + + # options for plugins + plugin_opts.add_argument( + "--nvd-api-key", + dest="nvd_api_key", + help="API key for NVD API", + metavar="NVDAPIKEY", + ) + plugin_opts.add_argument( + "--cve-db-predownload", + dest="cve_db_predownload", + action="store_true", + help="Enable CVE database predownload.URL should be defined by CVE_DB_PREDOWNLOAD_URL in conf/local.conf.", + ) + plugin_opts.add_argument( + "--update-cve-databese-only", + dest="update_cve_databese_only", + default=False, + action="store_true", + help="Do not run cve check. Update CVE database only.", + ) + plugin_opts.add_argument( + "--skip-update", + default=False, + action="store_true", + help="Skip update CVE databases", + ) + plugin_opts.add_argument( + "--disable-plugins", + help="List plugin names to be disabled without .py extension (comma separated). e.g. --disable-plugins eml_cve_debian_plugin,eml_cve_your_plugin", + ) return parser.parse_args() + if __name__ == "__main__": main(parse_options()) diff --git a/scripts/cve_check_ng.py b/scripts/cve_check_ng.py deleted file mode 100755 index 85fb9138..00000000 --- a/scripts/cve_check_ng.py +++ /dev/null @@ -1,494 +0,0 @@ -#!/usr/bin/python3 - -# -# EMLinux CVE checker -# -# Copyright (c) Cybertrust Japan Co., Ltd. -# -# SPDX-License-Identifier: MIT -# - -from lib.python.cve.plugin.eml_cve_plugin_base import EmlCvePlugin -import lib.python.cve.common_libs as cl -from lib.python.cve.nvd_lib import CveCheckMergedList, NvdCveInfoListCreator -from lib.python.cve.cve_reporter import CveReporter -from lib.python.package_info import PackageInfoHelper, PackageList -from lib.python.cve.cve_product import CveProductList -from lib.python.cve.kev_info import KevInfoList -import lib.python.cve.kev_cve as kev_cve -import lib.python.bitbake_runner as bitbake_runner - -import argparse -import sys -import os, os.path -import yaml -from typing import Any -import pathlib -import traceback -import re -import logging - -logging.basicConfig(level=logging.INFO, format="%(asctime)s:%(levelname)s: %(message)s") -logger = logging.getLogger("emlinux-cve-check") - -import glob - -import importlib.util -from concurrent.futures import ThreadPoolExecutor, as_completed - - -def create_ignore_list( - emlinux_layer_dir: str, - installed_packages: PackageList, - debian_codename: str, - extra_cve_check_ignore: str, -): - name = f"{emlinux_layer_dir}/conf/cve/cve_check_ignore.yml" - - def read_ignore_list(filename): - if filename is None: - return {} - - try: - with open(filename) as f: - return yaml.safe_load(f) - except FileNotFoundError: - logger.warn(f"File {filename} is not found") - return {} - - # 1. Read default ignore data - tmp_merge_list = read_ignore_list(name) - # 2. Read extra ignore list - extra_data = read_ignore_list(extra_cve_check_ignore) - - # backward compatibility - def make_backward_compatibility(data): - tmp = {} - for pkg in data: - if type(data[pkg]) is list: - tmp[pkg] = {"all": data[pkg]} - else: - tmp[pkg] = data[pkg] - return tmp - - tmp_merge_list = make_backward_compatibility(tmp_merge_list) - extra_data = make_backward_compatibility(extra_data) - - # 3. Merge default and extra list - if extra_data: - for pkg in extra_data: - if pkg in tmp_merge_list: - for ek in extra_data[pkg].keys(): - if ek in tmp_merge_list[pkg]: - tmp_merge_list[pkg][ek] += extra_data[pkg][ek] - else: - tmp_merge_list[pkg][ek] = extra_data[pkg][ek] - else: - tmp_merge_list[pkg] = extra_data[pkg] - - # 4. Create complete ignore list - # This step does not collect CVE IDs which are not target distribution/linux version. - ignore_list = {} - all_used = [] - for pkg in tmp_merge_list: - d = tmp_merge_list[pkg] - for version in d.keys(): - if debian_codename == version: - ignore_list[pkg] = d[version] - elif version == "all": - if not pkg in ignore_list: - ignore_list[pkg] = [] - all_used.append(pkg) - else: - if not pkg in installed_packages: - continue - pkg_version = str(installed_packages.get_upstream_version(pkg)) - if str(version) == pkg_version: - ignore_list[pkg] = d[version] - break - - ver_pattern = rf"\b{re.escape(str(version))}(?!\d)" - m = re.search(ver_pattern, pkg_version) - if m: - if not pkg in ignore_list: - ignore_list[pkg] = d[version] - else: - ignore_list[pkg].extend(d[version]) - - for pkg in all_used: - ignore_list[pkg].extend(tmp_merge_list[pkg]["all"]) - return ignore_list - - -def create_cve_check_merged_list( - src_pkg_names: list[str], check_results: Any -) -> CveCheckMergedList: - cve_check_merged_list = CveCheckMergedList() - - cve_ids = create_cve_id_list_by_src_pkg_name_from_check_result( - src_pkg_names, check_results - ) - for src_pkg_name in src_pkg_names: - for cveid in cve_ids: - for cr in check_results: - vulns = cr[src_pkg_name] - if vulns is None: - # Plugin doesn't have CVE information for the src_pkg_name - continue - if cveid in vulns: - cve_check_merged_list.add_data( - src_pkg_name, cveid, vulns[cveid], cr.priority - ) - else: - # Plugin doesn't have CVE information for the CVE - pass - # logger.debug(f"{cveid} {src_pkg_name} is not found") - return cve_check_merged_list - - -def create_cve_id_list_by_src_pkg_name_from_check_result( - src_pkg_names: list[str], check_results: Any -) -> list[str]: - tmp_cve_ids = [] - for src_pkg_name in src_pkg_names: - for cr in check_results: - ci = cr.cve_ids_by_src_pkg(src_pkg_name) - if ci: - tmp_cve_ids.extend(ci) - return list(dict.fromkeys(tmp_cve_ids)) - - -# make a source package name list which content has CVE information -def create_src_package_name_list_from_check_result(check_results: Any) -> list[str]: - tmp_list = [] - for result in check_results: - tmp_list.extend(result.src_pkg_names()) - return list(dict.fromkeys(tmp_list)) - - -def read_recipe_source_info(deploy_dir: str) -> Any: - filepath = deploy_dir + "/all-source-info.json" - return cl.read_json(filepath) - - -def load_plugin(plugin_file: str) -> EmlCvePlugin: - path = pathlib.Path(plugin_file).resolve() - if not path.exists(): - logger.error(f"Failed to find plugin {plugin_file}") - exit(1) - - spec = importlib.util.spec_from_file_location(path.stem, str(path)) - if not spec or not spec.loader: - logger.error(f"Fail to load spec from {plugin_file}") - exit(1) - - mod = importlib.util.module_from_spec(spec) - sys.modules[path.stem] = mod - try: - spec.loader.exec_module(mod) - except Exception as e: - logger.error(e) - traceback.print_exc() - exit(1) - - for obj in vars(mod).values(): - if ( - isinstance(obj, type) - and issubclass(obj, EmlCvePlugin) - and obj is not EmlCvePlugin - ): - return obj - - logger.info(f"Plugin is not found in {plugin_file}") - - return None - - -def load_plugins(plugin_files: list[str]): - plugins = [] - - for plugin_file in plugin_files: - logger.debug(f"loading {plugin_file}") - obj = load_plugin(plugin_file) - if obj: - plugins.append(obj) - - return plugins - - -# Plugin file name convention: -# 1. Plugin file must be located in scripts/lib/python/cve/plugin directory -# 2. Plugin file name must be start with eml_cve_ then ends with _plugin.py -# e.g. eml_cve_myplugin_plugin.py -def find_plugins(disable_plugins: list[str]) -> list[str]: - layer_dirs = bitbake_runner.find_layers() - plugins = [] - - plugin_dir = "/scripts/lib/python/cve/plugin/" - for ld in layer_dirs: - d = ld + plugin_dir - pattern = f"{d}/eml_cve_*_plugin.py" - for plugin in glob.glob(pattern): - p = os.path.splitext(os.path.basename(plugin))[0] - if not p in disable_plugins: - plugins.append(plugin) - else: - logger.info(f"Plugin '{p}' is disabled") - - return plugins - - -def cve_check_worker(plugin: EmlCvePlugin, args: Any): - logger.debug(f"run {plugin.plugin_name}") - - if not args.skip_update: - ret = plugin.update_database() - if not ret: - raise Exception(f"{plugin.plugin_name}: Failed to update datebase") - - if args.update_cve_databese_only: - return {} - - return plugin.run_check() - - -def fetch_kev_data(cve_data_dir: str) -> KevInfoList: - try: - kev_json = kev_cve.fetch_kev_data(cve_data_dir) - return KevInfoList(cl.read_json(kev_json)) - except: - return KevInfoList({}) - - -def create_disable_plugins_list(user_given_plugins: str) -> list[str]: - if not user_given_plugins: - return [] - - disable_plugins = [] - - do_not_disable = ["eml_cve_nvd_plugin"] - disable_plugins_tmp = [p.strip() for p in user_given_plugins.split(",")] - for p in disable_plugins_tmp: - if p not in do_not_disable: - disable_plugins.append(p) - else: - logger.warning(f"Plugin '{p}' cannot be disabeld") - - return disable_plugins - - -def main(args: dict): - if args.verbose_output: - logger.setLevel(logging.DEBUG) - - bitbakeinfo = bitbake_runner.get_bitbake_information(args.image_name) - - disable_plugins = create_disable_plugins_list(args.disable_plugins) - - dpkg_status_file = ( - bitbakeinfo["dpkg_status"] - if not args.dpkg_status_file - else args.dpkg_status_file - ) - - debian_codename = args.debian_codename - if not debian_codename: - debian_codename = bitbakeinfo["image_distro"].split("-")[1] - - installed_packages = None - cve_product_list = None - - if args.update_cve_databese_only: - logger.info("Run on database update only mode.") - else: - if not os.path.exists(dpkg_status_file): - logger.error(f"File {dpkg_status_file} is not found.") - exit(1) - - # Read dpkg file to get installed package information - installed_packages = PackageInfoHelper.parse_dpkg_status_file( - dpkg_status_file, - debian_codename, - target_source_package=args.target_source_package, - ) - - # Check recipe's source code provenance - recipe_source_info = read_recipe_source_info(bitbakeinfo["deploy_image_dir"]) - installed_packages.merge_recipe_source_info(recipe_source_info) - - # Read cve product list - cve_product_list = CveProductList() - cve_product_list.create_product_list( - installed_packages, bitbakeinfo["emlinux_layer_dir"], args.extra_cve_product - ) - - # Read ignore list - ignore_list = create_ignore_list( - bitbakeinfo["emlinux_layer_dir"], - installed_packages, - debian_codename, - args.extra_cve_check_ignore, - ) - - cve_data_dir = f"{bitbakeinfo['dl_dir']}/CVE" - cl.create_directory(cve_data_dir) - - # Find and load plugins - plugin_files = find_plugins(disable_plugins) - plugin_objs = load_plugins(plugin_files) - - # Create plugin instance - plugins = [] - for obj in plugin_objs: - o = obj(cve_data_dir, args, bitbakeinfo, installed_packages, cve_product_list) - plugins.append(o) - - check_results = [] - max_workers = min(args.threads, len(plugins)) - - # Run all plugins - with ThreadPoolExecutor(max_workers=max_workers) as ex: - futures = [ex.submit(cve_check_worker, p, args) for p in plugins] - for f in as_completed(futures): - try: - check_results.append(f.result()) - except Exception as e: - logger.error(f"error: {e}") - traceback.print_exc() - exit(1) - - if args.update_cve_databese_only: - logger.info("Updating database finished.") - exit(0) - - # Sort CVE data by plugin priority - check_results = sorted(check_results, key=lambda d: d.priority) - - # Merge CVE results - src_pkg_names = create_src_package_name_list_from_check_result(check_results) - - cve_check_merged_list = create_cve_check_merged_list(src_pkg_names, check_results) - - cve_check_merged_list.apply_ignore_list_info(ignore_list) - - # Load KEV data - kev_info_list = fetch_kev_data(cve_data_dir) - - # Create CVE report data - creator = NvdCveInfoListCreator(cve_data_dir, installed_packages, kev_info_list) - creator.create_cve_info_list(cve_check_merged_list) - - cve_info_list = creator.get_nvd_info_list() - - # Write CVE report - output_base_dir = ( - f"{bitbakeinfo['deploy_dir']}/cve/{bitbakeinfo['image_full_name']}" - ) - # Use cve_check_ng scripts own directory for testing - output_base_dir = f"{output_base_dir}/cve_check_ng" - - reporter = CveReporter(output_base_dir, bitbakeinfo["image_full_name"]) - reporter.write_report(args.output_format, cve_info_list, installed_packages) - - -def parse_options(): - parser = argparse.ArgumentParser() - plugin_opts = parser.add_argument_group("arguments for plugins") - cve_check_opts = parser.add_argument_group("arguments for cve check") - - # misc options - parser.add_argument( - "--verbose", - dest="verbose_output", - help="Enable verbose output", - default=False, - action="store_true", - ) - - # CVE check core options - cve_check_opts.add_argument( - "--debian-codename", - dest="debian_codename", - help="debian codename(bookworm, trixie, and etc)", - metavar="DEBIANCODENAME", - ) - cve_check_opts.add_argument( - "--output-format", - dest="output_format", - help="output format. available formats are text, json. formats can be comma separated string(e.g. text,json)", - default="text", - metavar="OUTPUTFORMAT", - ) - cve_check_opts.add_argument( - "--cve-product", - dest="extra_cve_product", - help="User defined cve-product file", - metavar="CVEPRODUCT", - ) - cve_check_opts.add_argument( - "--cve-ignore", - dest="extra_cve_check_ignore", - help="User defined cve-check-ignore file", - metavar="CVEIGNORE", - ) - cve_check_opts.add_argument( - "--image-name", - dest="image_name", - help="EMLinux image name(e.g. emlinux-image-base, emlinux-image-weston)", - metavar="IMAGENAME", - required=True, - ) - cve_check_opts.add_argument( - "--target-source-package", - dest="target_source_package", - help="Only check given debian source package(e.g. bash, util-linux", - metavar="DEBIAN SOURCE PACKAGE NAME", - ) - cve_check_opts.add_argument( - "--dpkg-status-file", - dest="dpkg_status_file", - help="Use specific dpkg_status file instead of default", - metavar="DPKG STATUS FILE", - ) - cve_check_opts.add_argument( - "--threads", default=1, help="Number of thread for cve check" - ) - - # options for plugins - plugin_opts.add_argument( - "--nvd-api-key", - dest="nvd_api_key", - help="API key for NVD API", - metavar="NVDAPIKEY", - ) - plugin_opts.add_argument( - "--cve-db-predownload", - dest="cve_db_predownload", - action="store_true", - help="Enable CVE database predownload.URL should be defined by CVE_DB_PREDOWNLOAD_URL in conf/local.conf.", - ) - plugin_opts.add_argument( - "--update-cve-databese-only", - dest="update_cve_databese_only", - default=False, - action="store_true", - help="Do not run cve check. Update CVE database only.", - ) - plugin_opts.add_argument( - "--skip-update", - default=False, - action="store_true", - help="Skip update CVE databases", - ) - plugin_opts.add_argument( - "--disable-plugins", - help="List plugin names to be disabled without .py extension (comma separated). e.g. --disable-plugins eml_cve_debian_plugin,eml_cve_your_plugin", - ) - return parser.parse_args() - - -if __name__ == "__main__": - logger.info("|------------------------------|") - logger.info("| This is experimental version |") - logger.info("|------------------------------|") - main(parse_options()) diff --git a/scripts/lib/python/cve/debian_cve.py b/scripts/lib/python/cve/debian_cve.py deleted file mode 100644 index 3e568b53..00000000 --- a/scripts/lib/python/cve/debian_cve.py +++ /dev/null @@ -1,124 +0,0 @@ -# -# EMLinux CVE checker. -# Download and store Debian's CVE data -# -# Copyright (c) Cybertrust Japan Co., Ltd. -# -# SPDX-License-Identifier: MIT -# - -import urllib.request -import gzip -import json -import os.path -import time -import debian.debian_support -import logging - -logger = logging.getLogger("emlinux-cve-check") -DEBIAN_CVE_TRACKER_JSON_URL = "https://security-tracker.debian.org/tracker/data/json" - -def remove_extra_suffix_in_version_string(version_str): - vs = version_str.strip() - - # Some version string in NVD database contains additional suffix - # such as _ubuntu1, _exp, and so of - if "\\" in vs: - vs = vs.split("\\")[0] - if "_" in vs: - vs = vs.split("_")[0] - if "+" in vs: - vs = vs.split("+")[0] - - return vs - -def parse_version(version_str): - vs = remove_extra_suffix_in_version_string(version_str) - - v = debian.debian_support.Version(vs) - - # Some debian packages added extra version info to upstream version. - if "+" in v.upstream_version: - v.upstream_version = v.upstream_version.split("+")[0] - if ".dfsg" in v.upstream_version: - v.upstream_version = v.upstream_version.split(".dfsg")[0] - - return v - -def parse_dpkg_status(dpkgstatus): - ret = [] - - with open(dpkgstatus, "r") as f: - lines = f.readlines() - d = {} - for line in lines: - line = line.strip() - if line.startswith("Package:"): - d["package"] = line.split(":")[1].strip() - elif line.startswith("Source"): - # some package contain version number so remove it. - # util-linux (2.38.1-5) - d["source"] = line.split(":")[1].strip().split(" ")[0].strip() - elif line.startswith("Version"): - tmp = line.split(" ")[1].strip() - d["version"] = debian.debian_support.Version(tmp) - v = parse_version(tmp) - d["upstream_version"] = parse_version(v.upstream_version) - elif len(line) == 0: - if not "source" in d: - # If source is not found in data, source package name should - # be same as binary package name - d["source"] = d["package"] - - ret.append(d) - d = {} - - return ret - -def fetch_json_data(): - request = urllib.request.Request(DEBIAN_CVE_TRACKER_JSON_URL) - for attempt in range(5): - try: - r = urllib.request.urlopen(request) - - if (r.headers['content-encoding'] == 'gzip'): - buf = r.read() - raw_data = gzip.decompress(buf) - else: - raw_data = r.read().decode("utf-8") - - r.close() - except Exception as e: - logger.debug(f"json file: received error ({e}), retrying") - time.sleep(6) - pass - else: - return json.loads(raw_data) - else: - # We failed at all attempts - return None - -def is_skip_fetch_json_file(json_file): - if json_file: - if os.path.exists(json_file): - if time.time() - os.path.getmtime(json_file) < 86400: - logger.info(f"Last database update is in 1day so skip Debian CVE database update") - return True - - return False - -def fetch_cve_data(dl_dir): - logger.info("Update debian CVE database") - debian_cve_json = f"{dl_dir}/debian_cves.json" - if is_skip_fetch_json_file(debian_cve_json): - return debian_cve_json - - data = fetch_json_data() - if data is None: - return None - - with open(debian_cve_json, "w") as f: - json.dump(data, f) - - return debian_cve_json - diff --git a/scripts/lib/python/cve/kernel_cve.py b/scripts/lib/python/cve/kernel_cve.py deleted file mode 100644 index 54c1c473..00000000 --- a/scripts/lib/python/cve/kernel_cve.py +++ /dev/null @@ -1,126 +0,0 @@ -# -# EMLinux CVE checker. -# Download cip-kernel-sec -# -# Copyright (c) Cybertrust Japan Co., Ltd. -# -# SPDX-License-Identifier: MIT -# - -import os, os.path -import subprocess -import sys -import shutil -import time -import yaml -import logging -import tempfile - -logger = logging.getLogger("emlinux-cve-check") - -def update_remote(remotes_path): - with open(remotes_path) as f: - content = yaml.safe_load(f) - - with open(remotes_path, "w") as f: - yaml.dump({"cip": content["cip"]}, f, default_flow_style=False) - -def run_cip_kernel_sec(kernel_src_dir, kver, cip_kernel_sec_dir): - cwd =os.getcwd() - cves = { - "Patched": [], - "Unpatched": [], - } - - os.chdir(cip_kernel_sec_dir) - - if not kver.startswith("v"): - kver = f"v{kver}" - - with tempfile.NamedTemporaryFile(delete=False) as f: - output_filename = f.name - - retcode = -1 - cmd = ["./scripts/report_affected.py", "--include-fixed", "--output-format=yaml", - f"--output-filename={output_filename}", "--git-repo", - kernel_src_dir, "--remote-name", "cip:origin", "--include-ignored", kver] - with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc: - proc.wait() - retcode = int(proc.returncode) - - if not retcode == 0: - logger.warning('Failed to run cip-kernel-sec') - for s in proc.stderr: - logger.warning(s.decode()) - - with open(output_filename) as f: - yaml_data = yaml.safe_load(f.read()) - k = list(yaml_data)[0] - - cves["Patched"] = yaml_data[k]["fixed"] - cves["Unpatched"] = yaml_data[k]["affected"] - - if retcode == 0: - os.unlink(output_filename) - - os.chdir(cwd) - - return cves - -def clone_cip_kernel_sec(): - logger.info("clone cip-kernel-sec") - git_uri = "https://gitlab.com/cip-project/cip-kernel/cip-kernel-sec.git" - - cmd = [ 'git', 'clone', git_uri ] - with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc: - proc.wait() - retcode = int(proc.returncode) - - if not retcode == 0: - logger.warning('Failed to clone cip-kernel-sec') - return False - - remotes_path = "./cip-kernel-sec/conf/remotes.yml" - update_remote(remotes_path) - - return True - -def update_cip_kernel_sec(): - logger.info("Update cip-kernel-sec") - - cmd = [ 'git', 'pull' ] - with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc: - proc.wait() - retcode = int(proc.returncode) - - if not retcode == 0: - logger.warning('Failed to pull cip-kernel-sec') - return False - - return True - - -def fetch_cip_kernel_sec(dl_dir): - cip_kernel_sec_dir = f"{dl_dir}/cip-kernel-sec" - - cwd = os.getcwd() - - need_clone = False - if not os.path.exists(cip_kernel_sec_dir): - need_clone = True - - if need_clone: - os.chdir(dl_dir) - ret = clone_cip_kernel_sec() - else: - os.chdir(cip_kernel_sec_dir) - ret = update_cip_kernel_sec() - - os.chdir(cwd) - - if not ret: - # Remove old cip-kernel-sec directory then clone all data next time. - shutil.rmtree(clone_cip_kernel_sec) - return None - - return cip_kernel_sec_dir diff --git a/scripts/lib/python/cve/nvd_cve.py b/scripts/lib/python/cve/nvd_cve.py deleted file mode 100644 index 0558cbd6..00000000 --- a/scripts/lib/python/cve/nvd_cve.py +++ /dev/null @@ -1,353 +0,0 @@ -# -# EMLinux CVE checker. -# Download and store NVD's CVE data -# -# Copyright (c) Cybertrust Japan Co., Ltd. -# -# SPDX-License-Identifier: MIT -# - -import sqlite3 -import datetime -import urllib.request -import urllib.parse -import gzip -import http -import time -import json -import logging - -CVE_DATABASE_NAME = "nvd_cve_db.db" - -CVE_DB_UPDATE_INTERVAL = 86400 - -NVDCVE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0" -NVDCVE_API_KEY = None - -logger = logging.getLogger("emlinux-cve-check") - -def parse_node_and_insert(conn, node, cveId): - - def cpe_generator(): - for cpe in node.get('cpeMatch', ()): - if not cpe['vulnerable']: - return - cpe23 = cpe.get('criteria') - if not cpe23: - return - cpe23 = cpe23.split(':') - if len(cpe23) < 6: - return - vendor = cpe23[3] - product = cpe23[4] - version = cpe23[5] - - if cpe23[6] == '*' or cpe23[6] == '-': - version_suffix = "" - else: - version_suffix = "_" + cpe23[6] - - if version != '*' and version != '-': - # Version is defined, this is a '=' match - yield [cveId, vendor, product, version + version_suffix, '=', '', ''] - elif version == '-': - # no version information is available - yield [cveId, vendor, product, version, '', '', ''] - else: - # Parse start version, end version and operators - op_start = '' - op_end = '' - v_start = '' - v_end = '' - - if 'versionStartIncluding' in cpe: - op_start = '>=' - v_start = cpe['versionStartIncluding'] - - if 'versionStartExcluding' in cpe: - op_start = '>' - v_start = cpe['versionStartExcluding'] - - if 'versionEndIncluding' in cpe: - op_end = '<=' - v_end = cpe['versionEndIncluding'] - - if 'versionEndExcluding' in cpe: - op_end = '<' - v_end = cpe['versionEndExcluding'] - - if op_start or op_end or v_start or v_end: - yield [cveId, vendor, product, v_start, op_start, v_end, op_end] - else: - # This is no version information, expressed differently. - # Save processing by representing as -. - yield [cveId, vendor, product, '-', '', '', ''] - - conn.executemany("insert into PRODUCTS values (?, ?, ?, ?, ?, ?, ?)", cpe_generator()).close() - -def update_db(conn, elt): - """ - Update a single entry in the on-disk database - """ - - accessVector = None - vectorString = None - cveId = elt['cve']['id'] - logger.debug(f"Processing CVE {cveId}") - - if 'vulnStatus' in elt['cve']: - vulnStatus = elt['cve']['vulnStatus'] - else: - vulnStatus = "" - - cveDesc = "" - for desc in elt['cve']['descriptions']: - if desc['lang'] == 'en': - cveDesc = desc['value'] - date = elt['cve']['lastModified'] - try: - accessVector = elt['cve']['metrics']['cvssMetricV2'][0]['cvssData']['accessVector'] - vectorString = elt['cve']['metrics']['cvssMetricV2'][0]['cvssData']['vectorString'] - cvssv2 = elt['cve']['metrics']['cvssMetricV2'][0]['cvssData']['baseScore'] - except KeyError: - cvssv2 = 0.0 - cvssv3 = None - try: - accessVector = accessVector or elt['cve']['metrics']['cvssMetricV30'][0]['cvssData']['attackVector'] - vectorString = vectorString or elt['cve']['metrics']['cvssMetricV30'][0]['cvssData']['vectorString'] - cvssv3 = elt['cve']['metrics']['cvssMetricV30'][0]['cvssData']['baseScore'] - except KeyError: - pass - try: - accessVector = accessVector or elt['cve']['metrics']['cvssMetricV31'][0]['cvssData']['attackVector'] - vectorString = vectorString or elt['cve']['metrics']['cvssMetricV31'][0]['cvssData']['vectorString'] - cvssv3 = cvssv3 or elt['cve']['metrics']['cvssMetricV31'][0]['cvssData']['baseScore'] - except KeyError: - pass - accessVector = accessVector or "UNKNOWN" - vectorString = vectorString or "UNKNOWN" - cvssv3 = cvssv3 or 0.0 - - conn.execute("insert or replace into NVD values (?, ?, ?, ?, ?, ?, ?, ?)", - [cveId, vulnStatus, cveDesc, cvssv2, cvssv3, date, accessVector, vectorString]).close() - - try: - # Remove any pre-existing CVE configuration. Even for partial database - # update, those will be repopulated. This ensures that old - # configuration is not kept for an updated CVE. - conn.execute("delete from PRODUCTS where ID = ?", [cveId]).close() - for config in elt['cve']['configurations']: - # This is suboptimal as it doesn't handle AND/OR and negate, but is better than nothing - for node in config["nodes"]: - parse_node_and_insert(conn, node, cveId) - except KeyError: - logger.debug("CVE %s has no configurations" % cveId) - -def nvd_request_next(url, api_key, args): - """ - Request next part of the NVD dabase - """ - - request = urllib.request.Request(url + "?" + urllib.parse.urlencode(args)) - if api_key: - request.add_header("apiKey", api_key) - logger.debug(f"Requesting {request.full_url}") - - for attempt in range(5): - try: - r = urllib.request.urlopen(request) - - if (r.headers['content-encoding'] == 'gzip'): - buf = r.read() - raw_data = gzip.decompress(buf) - else: - raw_data = r.read().decode("utf-8") - - r.close() - except Exception as e: - logger.debug(f"CVE database: received error ({e}), retrying") - time.sleep(6) - pass - else: - return raw_data - else: - # We failed at all attempts - return None - -def fetch_all_cves(db_file, conn, last_modified, api_key): - - index = 0 - url = NVDCVE_URL - - req_args = {} - - if last_modified is not None: - req_args["lastModStartDate"] = last_modified - req_args["lastModEndDate"] = datetime.datetime.now().isoformat() - - # Recommended by NVD - sleep_time = 6 - if api_key: - sleep_time = 2 - - with open(db_file, 'a') as cve_f: - while True: - logger.debug("Updating entries") - - req_args["startIndex"] = index - - raw_data = nvd_request_next(url, api_key, req_args) - if raw_data is None: - return False - - data = json.loads(raw_data) - - index = data["startIndex"] - total = data["totalResults"] - per_page = data["resultsPerPage"] - logger.debug(f"Got {per_page} entries") - for cve in data["vulnerabilities"]: - update_db(conn, cve) - - if per_page == 0: - # no more data - break - - index += per_page - if index >= total: - break - - time.sleep(sleep_time) - - return True - -def update_last_modified_date(conn): - d = datetime.datetime.now().isoformat() - - with conn: - c = conn.cursor() - - cursor = c.execute("SELECT LASTMODIFIED from META where ID=1") - last = cursor.fetchone() - - if last is None: - sql = f"INSERT INTO META VALUES (1, '{d}')" - else: - sql = f"UPDATE META set LASTMODIFIED='{d}' where ID=1" - - c.execute(sql) - - c.close() - - -def get_last_modified_date(conn): - with conn: - c = conn.cursor() - - cursor = c.execute("SELECT LASTMODIFIED from META") - - last = cursor.fetchone() - c.close() - - if last is None: - return None - - return last[0] - -def initialize_nvd_cve_db(conn): - with conn: - c = conn.cursor() - - c.execute("CREATE TABLE IF NOT EXISTS META (ID NUMBER UNIQUE, LASTMODIFIED TEXT)") - - c.execute("CREATE TABLE IF NOT EXISTS NVD (ID TEXT UNIQUE, VULNSTATUS TEXT, SUMMARY TEXT, SCOREV2 TEXT, \ - SCOREV3 TEXT, MODIFIED INTEGER, VECTOR TEXT, VECTORSTRING TEXT)") - - c.execute("CREATE TABLE IF NOT EXISTS PRODUCTS (ID TEXT, \ - VENDOR TEXT, PRODUCT TEXT, VERSION_START TEXT, OPERATOR_START TEXT, \ - VERSION_END TEXT, OPERATOR_END TEXT)") - - c.execute("CREATE INDEX IF NOT EXISTS PRODUCT_ID_IDX on PRODUCTS(ID);") - - c.close() - -def predownload_db(predownload_url, db_file): - logger.info(f"Download CVE database file from {predownload_url}.") - - request = urllib.request.Request(predownload_url) - for attempt in range(5): - try: - r = urllib.request.urlopen(request) - - if (r.headers['content-encoding'] == 'gzip'): - buf = r.read() - raw_data = gzip.decompress(buf) - else: - raw_data = r.read() - - r.close() - except Exception as e: - logger.debug(f"CVE databese download: received error ({e}), retrying") - time.sleep(6) - pass - else: - with open(db_file, "wb") as f: - f.write(raw_data) - logger.info(f"Download CVE database file was succeeded.") - - return True - else: - # We failed at all attempts - return False - - -def check_skip_db_update(conn): - skip_db_update = False - last_modified = get_last_modified_date(conn) - if last_modified: - d1 = datetime.datetime.fromisoformat(datetime.datetime.now().isoformat()) - d2 = datetime.datetime.fromisoformat(last_modified) - - date_delta = d1 - d2 - if date_delta.total_seconds() < CVE_DB_UPDATE_INTERVAL: - skip_db_update = True - else: - # Database is too old so that fetch all data - if date_delta.days > 120: - last_modified = None - - return skip_db_update, last_modified - -def update_nvd_db(dl_dir, nvd_api_key, predownload_url): - db_file = f"{dl_dir}/{CVE_DATABASE_NAME}" - result = False - - conn = sqlite3.connect(db_file) - logger.debug(f"Initialize nvd cve database {db_file}") - initialize_nvd_cve_db(conn) - - skip_db_update, last_modified = check_skip_db_update(conn) - - if not skip_db_update and predownload_url: - # predownload database file is old, download latest file - conn.close() - logger.info("Predownload CVE database file.") - if not predownload_db(predownload_url, db_file): - return result, None - conn = sqlite3.connect(db_file) - # re-check last modified date - skip_db_update, last_modified = check_skip_db_update(conn) - - if not skip_db_update: - logger.info("Update NVD CVE database") - if fetch_all_cves(db_file, conn, last_modified, nvd_api_key): - logger.info("Update last modified date") - update_last_modified_date(conn) - conn.commit() - result = True - else: - logger.info(f"Last database update is in 1day so skip NVD database update") - result = True - - conn.close() - return result, db_file