From 80672063d5d8a1e47bd241c60f23f71ecb9ce7ad Mon Sep 17 00:00:00 2001 From: Masami Ichikawa Date: Mon, 10 Aug 2026 04:58:47 +0000 Subject: [PATCH 1/2] cve-check: Remove cve_check.py and related code This commit removed cve_check.py and its related files to prepare replacing cve_check_ng.py to rename cve_check.py. Signed-off-by: Masami Ichikawa --- scripts/cve_check.py | 781 --------------------------- scripts/lib/python/cve/debian_cve.py | 124 ----- scripts/lib/python/cve/kernel_cve.py | 126 ----- scripts/lib/python/cve/nvd_cve.py | 353 ------------ 4 files changed, 1384 deletions(-) delete mode 100755 scripts/cve_check.py delete mode 100644 scripts/lib/python/cve/debian_cve.py delete mode 100644 scripts/lib/python/cve/kernel_cve.py delete mode 100644 scripts/lib/python/cve/nvd_cve.py diff --git a/scripts/cve_check.py b/scripts/cve_check.py deleted file mode 100755 index b4e1a4c1..00000000 --- a/scripts/cve_check.py +++ /dev/null @@ -1,781 +0,0 @@ -#!/usr/bin/python3 - -# -# EMLinux CVE checker -# -# Copyright (c) Cybertrust Japan Co., Ltd. -# -# SPDX-License-Identifier: MIT -# -import argparse -import sys -import os, os.path -import sqlite3 -import yaml -import logging -import debian.debian_support -import re - -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()) - -def get_cve_products(extra_cve_product): - name = os.path.join(os.path.dirname(__file__), "../conf/cve/cve_products.yml") - - 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 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 uniq_installed_pkgs: - continue - pkg_version = str(uniq_installed_pkgs[pkg]["upstream_version"]) - 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 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 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) - - return debian_pkg_cve_data, cve_not_found_pkgs - -def create_unique_package(installed): - ret = {} - for pkg in installed: - src = pkg['source'] - - 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 = {} - - bin_pkgname = uniq_installed_pkgs[pkgname]["bin_pkgs"] - - 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) - - 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) - - 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 - - return cves - -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 - - return kev_data - -def add_kev_info(cves, kev_list): - kev_data = create_kev_data(kev_list) - - 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"] - - return cves - -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 ...") - - 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) - - 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) - - 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 write_text(cves, output_dir, uniq_installed_pkgs): - filenames = [] - - for pkgname in cves: - if len(cves[pkgname]) == 0: - continue - - filename = f"{output_dir}/{pkgname}" - filenames.append(filename) - - 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 write_json(cves, output_dir, uniq_installed_pkgs, cve_products): - filenames = [] - - for pkgname in cves: - filename = f"{output_dir}/{pkgname}_cve.json" - filenames.append(filename) - - info = cves[pkgname] - pkginfo = uniq_installed_pkgs[pkgname] - - product = pkgname - if pkgname in cve_products: - product = cve_products[pkgname]["product"] - - 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) - - 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") - 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) - - if args.update_cve_databese_only: - logger.info("Finish CVE database update.") - 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) - - 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) - - 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) - - 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) - - 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) - - cves = update_ignored_cves_status(cves, cve_ignore_list) - - output_base_dir = f"{bitbakeinfo['deploy_dir']}/cve/{bitbakeinfo['image_full_name']}" - - create_directory(output_base_dir) - - 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) - - 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") - - return parser.parse_args() - -if __name__ == "__main__": - 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 From 789920c2fbe9eb44a570dabc2f399309e286ad58 Mon Sep 17 00:00:00 2001 From: Masami Ichikawa Date: Thu, 27 Aug 2026 05:21:19 +0000 Subject: [PATCH 2/2] cve-check: Make cve_check_ng.py as default cve check script It time to change cve_check_ng.py as default cve check script. To make cve_check_ng.py as default scripts, we updated following list. - Rename cve_check_ng.py to cve_check.py - Remove experimental messages - Change CVE report file directory Signed-off-by: Masami Ichikawa --- scripts/{cve_check_ng.py => cve_check.py} | 5 ----- 1 file changed, 5 deletions(-) rename scripts/{cve_check_ng.py => cve_check.py} (98%) diff --git a/scripts/cve_check_ng.py b/scripts/cve_check.py similarity index 98% rename from scripts/cve_check_ng.py rename to scripts/cve_check.py index 85fb9138..d0bf35e7 100755 --- a/scripts/cve_check_ng.py +++ b/scripts/cve_check.py @@ -384,8 +384,6 @@ def main(args: dict): 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) @@ -488,7 +486,4 @@ def parse_options(): if __name__ == "__main__": - logger.info("|------------------------------|") - logger.info("| This is experimental version |") - logger.info("|------------------------------|") main(parse_options())