e3ef352ef6
This commit modifies cve.py, as well as its users cve-checker and pkg-stats to support CPE ID based matching, for packages that have CPE ID information. One of the non-trivial thing is that we can't simply iterate over all CVEs, and then iterate over all our packages to see which packages have CPE ID information that match the CPEs affected by the CVE. Indeed, this is an O(n^2) operation. So instead, we do a pre-filtering of packages potentially affected. In check_package_cves(), we build a cpe_product_pkgs dict that associates a CPE product name to the packages that have this CPE product name. The CPE product name is either derived from the CPE information provided by the package if available, and otherwise we use the package name, which is what was used prior to this patch. And then, when we look at CVEs, we only consider the packages that have a CPE product name matching the CPE products affected by the CVEs. This is done in check_package_cve_affects(). Signed-off-by: Thomas Petazzoni <thomas.petazzoni@bootlin.com>
263 lines
9.1 KiB
Python
Executable File
263 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
# Copyright (C) 2009 by Thomas Petazzoni <thomas.petazzoni@free-electrons.com>
|
|
# Copyright (C) 2020 by Gregory CLEMENT <gregory.clement@bootlin.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
# General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software
|
|
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
|
|
import datetime
|
|
import os
|
|
import requests # URL checking
|
|
import distutils.version
|
|
import time
|
|
import gzip
|
|
import sys
|
|
import operator
|
|
|
|
try:
|
|
import ijson
|
|
except ImportError:
|
|
sys.stderr.write("You need ijson to parse NVD for CVE check\n")
|
|
exit(1)
|
|
|
|
sys.path.append('utils/')
|
|
|
|
NVD_START_YEAR = 2002
|
|
NVD_JSON_VERSION = "1.1"
|
|
NVD_BASE_URL = "https://nvd.nist.gov/feeds/json/cve/" + NVD_JSON_VERSION
|
|
|
|
ops = {
|
|
'>=': operator.ge,
|
|
'>': operator.gt,
|
|
'<=': operator.le,
|
|
'<': operator.lt,
|
|
'=': operator.eq
|
|
}
|
|
|
|
|
|
# Check if two CPE IDs match each other
|
|
def cpe_matches(cpe1, cpe2):
|
|
cpe1_elems = cpe1.split(":")
|
|
cpe2_elems = cpe2.split(":")
|
|
|
|
remains = filter(lambda x: x[0] not in ["*", "-"] and x[1] not in ["*", "-"] and x[0] != x[1],
|
|
zip(cpe1_elems, cpe2_elems))
|
|
return len(list(remains)) == 0
|
|
|
|
|
|
def cpe_product(cpe):
|
|
return cpe.split(':')[4]
|
|
|
|
|
|
def cpe_version(cpe):
|
|
return cpe.split(':')[5]
|
|
|
|
|
|
class CVE:
|
|
"""An accessor class for CVE Items in NVD files"""
|
|
CVE_AFFECTS = 1
|
|
CVE_DOESNT_AFFECT = 2
|
|
CVE_UNKNOWN = 3
|
|
|
|
def __init__(self, nvd_cve):
|
|
"""Initialize a CVE from its NVD JSON representation"""
|
|
self.nvd_cve = nvd_cve
|
|
|
|
@staticmethod
|
|
def download_nvd_year(nvd_path, year):
|
|
metaf = "nvdcve-%s-%s.meta" % (NVD_JSON_VERSION, year)
|
|
path_metaf = os.path.join(nvd_path, metaf)
|
|
jsonf_gz = "nvdcve-%s-%s.json.gz" % (NVD_JSON_VERSION, year)
|
|
path_jsonf_gz = os.path.join(nvd_path, jsonf_gz)
|
|
|
|
# If the database file is less than a day old, we assume the NVD data
|
|
# locally available is recent enough.
|
|
if os.path.exists(path_jsonf_gz) and os.stat(path_jsonf_gz).st_mtime >= time.time() - 86400:
|
|
return path_jsonf_gz
|
|
|
|
# If not, we download the meta file
|
|
url = "%s/%s" % (NVD_BASE_URL, metaf)
|
|
print("Getting %s" % url)
|
|
page_meta = requests.get(url)
|
|
page_meta.raise_for_status()
|
|
|
|
# If the meta file already existed, we compare the existing
|
|
# one with the data newly downloaded. If they are different,
|
|
# we need to re-download the database.
|
|
# If the database does not exist locally, we need to redownload it in
|
|
# any case.
|
|
if os.path.exists(path_metaf) and os.path.exists(path_jsonf_gz):
|
|
meta_known = open(path_metaf, "r").read()
|
|
if page_meta.text == meta_known:
|
|
return path_jsonf_gz
|
|
|
|
# Grab the compressed JSON NVD, and write files to disk
|
|
url = "%s/%s" % (NVD_BASE_URL, jsonf_gz)
|
|
print("Getting %s" % url)
|
|
page_json = requests.get(url)
|
|
page_json.raise_for_status()
|
|
open(path_jsonf_gz, "wb").write(page_json.content)
|
|
open(path_metaf, "w").write(page_meta.text)
|
|
return path_jsonf_gz
|
|
|
|
@classmethod
|
|
def read_nvd_dir(cls, nvd_dir):
|
|
"""
|
|
Iterate over all the CVEs contained in NIST Vulnerability Database
|
|
feeds since NVD_START_YEAR. If the files are missing or outdated in
|
|
nvd_dir, a fresh copy will be downloaded, and kept in .json.gz
|
|
"""
|
|
for year in range(NVD_START_YEAR, datetime.datetime.now().year + 1):
|
|
filename = CVE.download_nvd_year(nvd_dir, year)
|
|
try:
|
|
content = ijson.items(gzip.GzipFile(filename), 'CVE_Items.item')
|
|
except: # noqa: E722
|
|
print("ERROR: cannot read %s. Please remove the file then rerun this script" % filename)
|
|
raise
|
|
for cve in content:
|
|
yield cls(cve)
|
|
|
|
def each_product(self):
|
|
"""Iterate over each product section of this cve"""
|
|
for vendor in self.nvd_cve['cve']['affects']['vendor']['vendor_data']:
|
|
for product in vendor['product']['product_data']:
|
|
yield product
|
|
|
|
def parse_node(self, node):
|
|
"""
|
|
Parse the node inside the configurations section to extract the
|
|
cpe information usefull to know if a product is affected by
|
|
the CVE. Actually only the product name and the version
|
|
descriptor are needed, but we also provide the vendor name.
|
|
"""
|
|
|
|
# The node containing the cpe entries matching the CVE can also
|
|
# contain sub-nodes, so we need to manage it.
|
|
for child in node.get('children', ()):
|
|
for parsed_node in self.parse_node(child):
|
|
yield parsed_node
|
|
|
|
for cpe in node.get('cpe_match', ()):
|
|
if not cpe['vulnerable']:
|
|
return
|
|
product = cpe_product(cpe['cpe23Uri'])
|
|
version = cpe_version(cpe['cpe23Uri'])
|
|
# ignore when product is '-', which means N/A
|
|
if product == '-':
|
|
return
|
|
op_start = ''
|
|
op_end = ''
|
|
v_start = ''
|
|
v_end = ''
|
|
|
|
if version != '*' and version != '-':
|
|
# Version is defined, this is a '=' match
|
|
op_start = '='
|
|
v_start = version
|
|
else:
|
|
# Parse start version, end version and operators
|
|
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']
|
|
|
|
yield {
|
|
'id': cpe['cpe23Uri'],
|
|
'v_start': v_start,
|
|
'op_start': op_start,
|
|
'v_end': v_end,
|
|
'op_end': op_end
|
|
}
|
|
|
|
def each_cpe(self):
|
|
for node in self.nvd_cve['configurations']['nodes']:
|
|
for cpe in self.parse_node(node):
|
|
yield cpe
|
|
|
|
@property
|
|
def identifier(self):
|
|
"""The CVE unique identifier"""
|
|
return self.nvd_cve['cve']['CVE_data_meta']['ID']
|
|
|
|
@property
|
|
def affected_products(self):
|
|
"""The set of CPE products referred by this CVE definition"""
|
|
return set(cpe_product(p['id']) for p in self.each_cpe())
|
|
|
|
def affects(self, name, version, cve_ignore_list, cpeid=None):
|
|
"""
|
|
True if the Buildroot Package object passed as argument is affected
|
|
by this CVE.
|
|
"""
|
|
if self.identifier in cve_ignore_list:
|
|
return self.CVE_DOESNT_AFFECT
|
|
|
|
pkg_version = distutils.version.LooseVersion(version)
|
|
if not hasattr(pkg_version, "version"):
|
|
print("Cannot parse package '%s' version '%s'" % (name, version))
|
|
pkg_version = None
|
|
|
|
# if we don't have a cpeid, build one based on name and version
|
|
if not cpeid:
|
|
cpeid = "cpe:2.3:*:*:%s:%s:*:*:*:*:*:*:*" % (name, version)
|
|
|
|
for cpe in self.each_cpe():
|
|
if not cpe_matches(cpe['id'], cpeid):
|
|
continue
|
|
if not cpe['v_start'] and not cpe['v_end']:
|
|
return self.CVE_AFFECTS
|
|
if not pkg_version:
|
|
continue
|
|
|
|
if cpe['v_start']:
|
|
try:
|
|
cve_affected_version = distutils.version.LooseVersion(cpe['v_start'])
|
|
inrange = ops.get(cpe['op_start'])(pkg_version, cve_affected_version)
|
|
except TypeError:
|
|
return self.CVE_UNKNOWN
|
|
|
|
# current package version is before v_start, so we're
|
|
# not affected by the CVE
|
|
if not inrange:
|
|
continue
|
|
|
|
if cpe['v_end']:
|
|
try:
|
|
cve_affected_version = distutils.version.LooseVersion(cpe['v_end'])
|
|
inrange = ops.get(cpe['op_end'])(pkg_version, cve_affected_version)
|
|
except TypeError:
|
|
return self.CVE_UNKNOWN
|
|
|
|
# current package version is after v_end, so we're
|
|
# not affected by the CVE
|
|
if not inrange:
|
|
continue
|
|
|
|
# We're in the version range affected by this CVE
|
|
return self.CVE_AFFECTS
|
|
|
|
return self.CVE_DOESNT_AFFECT
|