a35f51cee1
The NVD files that are used to build the list of CVEs affecting Buildroot packages are quite large (a few hundreds MB of json), and cause the pkg-stats scripts to have a huge memory footprint (a few GB with Python 2.7). However, because we only need to iterate on CVE items one by one, we can process them in streaming (ie decoding one CVE at a time from the JSON representation). Because the json module from the python standard library does not support such a mode of operation, we switch to the third-party package ijson, which is compatible with both Python 2 and Python3. To run the script with these modifications, one should install the ijson python package. This can be done with pip: `pip install ijson`. On Debian based distributions, this can also be done with the apt package manager: `apt install python-ijson`. Signed-off-by: Titouan Christophe <titouan.christophe@railnova.eu> Reviewed-by: Thomas De Schampheleire <thomas.de_schampheleire@nokia.com> Tested-by: Thomas De Schampheleire <thomas.de_schampheleire@nokia.com> Signed-off-by: Peter Korsgaard <peter@korsgaard.com>
918 lines
31 KiB
Python
Executable File
918 lines
31 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
# Copyright (C) 2009 by Thomas Petazzoni <thomas.petazzoni@free-electrons.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 argparse
|
|
import datetime
|
|
import fnmatch
|
|
import os
|
|
from collections import defaultdict
|
|
import re
|
|
import subprocess
|
|
import requests # URL checking
|
|
import json
|
|
import ijson
|
|
import certifi
|
|
import distutils.version
|
|
import time
|
|
import gzip
|
|
from urllib3 import HTTPSConnectionPool
|
|
from urllib3.exceptions import HTTPError
|
|
from multiprocessing import Pool
|
|
|
|
NVD_START_YEAR = 2002
|
|
NVD_JSON_VERSION = "1.0"
|
|
NVD_BASE_URL = "https://nvd.nist.gov/feeds/json/cve/" + NVD_JSON_VERSION
|
|
|
|
INFRA_RE = re.compile(r"\$\(eval \$\(([a-z-]*)-package\)\)")
|
|
URL_RE = re.compile(r"\s*https?://\S*\s*$")
|
|
|
|
RM_API_STATUS_ERROR = 1
|
|
RM_API_STATUS_FOUND_BY_DISTRO = 2
|
|
RM_API_STATUS_FOUND_BY_PATTERN = 3
|
|
RM_API_STATUS_NOT_FOUND = 4
|
|
|
|
# Used to make multiple requests to the same host. It is global
|
|
# because it's used by sub-processes.
|
|
http_pool = None
|
|
|
|
|
|
class Package:
|
|
all_licenses = list()
|
|
all_license_files = list()
|
|
all_versions = dict()
|
|
all_ignored_cves = dict()
|
|
|
|
def __init__(self, name, path):
|
|
self.name = name
|
|
self.path = path
|
|
self.infras = None
|
|
self.has_license = False
|
|
self.has_license_files = False
|
|
self.has_hash = False
|
|
self.patch_count = 0
|
|
self.warnings = 0
|
|
self.current_version = None
|
|
self.url = None
|
|
self.url_status = None
|
|
self.url_worker = None
|
|
self.cves = list()
|
|
self.latest_version = (RM_API_STATUS_ERROR, None, None)
|
|
|
|
def pkgvar(self):
|
|
return self.name.upper().replace("-", "_")
|
|
|
|
def set_url(self):
|
|
"""
|
|
Fills in the .url field
|
|
"""
|
|
self.url_status = "No Config.in"
|
|
for filename in os.listdir(os.path.dirname(self.path)):
|
|
if fnmatch.fnmatch(filename, 'Config.*'):
|
|
fp = open(os.path.join(os.path.dirname(self.path), filename), "r")
|
|
for config_line in fp:
|
|
if URL_RE.match(config_line):
|
|
self.url = config_line.strip()
|
|
self.url_status = "Found"
|
|
fp.close()
|
|
return
|
|
self.url_status = "Missing"
|
|
fp.close()
|
|
|
|
def set_infra(self):
|
|
"""
|
|
Fills in the .infras field
|
|
"""
|
|
self.infras = list()
|
|
with open(self.path, 'r') as f:
|
|
lines = f.readlines()
|
|
for l in lines:
|
|
match = INFRA_RE.match(l)
|
|
if not match:
|
|
continue
|
|
infra = match.group(1)
|
|
if infra.startswith("host-"):
|
|
self.infras.append(("host", infra[5:]))
|
|
else:
|
|
self.infras.append(("target", infra))
|
|
|
|
def set_license(self):
|
|
"""
|
|
Fills in the .has_license and .has_license_files fields
|
|
"""
|
|
var = self.pkgvar()
|
|
if var in self.all_licenses:
|
|
self.has_license = True
|
|
if var in self.all_license_files:
|
|
self.has_license_files = True
|
|
|
|
def set_hash_info(self):
|
|
"""
|
|
Fills in the .has_hash field
|
|
"""
|
|
hashpath = self.path.replace(".mk", ".hash")
|
|
self.has_hash = os.path.exists(hashpath)
|
|
|
|
def set_patch_count(self):
|
|
"""
|
|
Fills in the .patch_count field
|
|
"""
|
|
self.patch_count = 0
|
|
pkgdir = os.path.dirname(self.path)
|
|
for subdir, _, _ in os.walk(pkgdir):
|
|
self.patch_count += len(fnmatch.filter(os.listdir(subdir), '*.patch'))
|
|
|
|
def set_current_version(self):
|
|
"""
|
|
Fills in the .current_version field
|
|
"""
|
|
var = self.pkgvar()
|
|
if var in self.all_versions:
|
|
self.current_version = self.all_versions[var]
|
|
|
|
def set_check_package_warnings(self):
|
|
"""
|
|
Fills in the .warnings field
|
|
"""
|
|
cmd = ["./utils/check-package"]
|
|
pkgdir = os.path.dirname(self.path)
|
|
for root, dirs, files in os.walk(pkgdir):
|
|
for f in files:
|
|
if f.endswith(".mk") or f.endswith(".hash") or f == "Config.in" or f == "Config.in.host":
|
|
cmd.append(os.path.join(root, f))
|
|
o = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()[1]
|
|
lines = o.splitlines()
|
|
for line in lines:
|
|
m = re.match("^([0-9]*) warnings generated", line)
|
|
if m:
|
|
self.warnings = int(m.group(1))
|
|
return
|
|
|
|
def is_cve_ignored(self, cve):
|
|
"""
|
|
Tells if the CVE is ignored by the package
|
|
"""
|
|
return cve in self.all_ignored_cves.get(self.pkgvar(), [])
|
|
|
|
def __eq__(self, other):
|
|
return self.path == other.path
|
|
|
|
def __lt__(self, other):
|
|
return self.path < other.path
|
|
|
|
def __str__(self):
|
|
return "%s (path='%s', license='%s', license_files='%s', hash='%s', patches=%d)" % \
|
|
(self.name, self.path, self.has_license, self.has_license_files, self.has_hash, self.patch_count)
|
|
|
|
|
|
class CVE:
|
|
"""An accessor class for CVE Items in NVD files"""
|
|
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:
|
|
print("ERROR: cannot read %s. Please remove the file then rerun this script" % filename)
|
|
raise
|
|
for cve in content:
|
|
yield cls(cve['cve'])
|
|
|
|
def each_product(self):
|
|
"""Iterate over each product section of this cve"""
|
|
for vendor in self.nvd_cve['affects']['vendor']['vendor_data']:
|
|
for product in vendor['product']['product_data']:
|
|
yield product
|
|
|
|
@property
|
|
def identifier(self):
|
|
"""The CVE unique identifier"""
|
|
return self.nvd_cve['CVE_data_meta']['ID']
|
|
|
|
@property
|
|
def pkg_names(self):
|
|
"""The set of package names referred by this CVE definition"""
|
|
return set(p['product_name'] for p in self.each_product())
|
|
|
|
def affects(self, br_pkg):
|
|
"""
|
|
True if the Buildroot Package object passed as argument is affected
|
|
by this CVE.
|
|
"""
|
|
for product in self.each_product():
|
|
if product['product_name'] != br_pkg.name:
|
|
continue
|
|
|
|
for v in product['version']['version_data']:
|
|
if v["version_affected"] == "=":
|
|
if br_pkg.current_version == v["version_value"]:
|
|
return True
|
|
elif v["version_affected"] == "<=":
|
|
pkg_version = distutils.version.LooseVersion(br_pkg.current_version)
|
|
if not hasattr(pkg_version, "version"):
|
|
print("Cannot parse package '%s' version '%s'" % (br_pkg.name, br_pkg.current_version))
|
|
continue
|
|
cve_affected_version = distutils.version.LooseVersion(v["version_value"])
|
|
if not hasattr(cve_affected_version, "version"):
|
|
print("Cannot parse CVE affected version '%s'" % v["version_value"])
|
|
continue
|
|
return pkg_version <= cve_affected_version
|
|
else:
|
|
print("version_affected: %s" % v['version_affected'])
|
|
return False
|
|
|
|
|
|
def get_pkglist(npackages, package_list):
|
|
"""
|
|
Builds the list of Buildroot packages, returning a list of Package
|
|
objects. Only the .name and .path fields of the Package object are
|
|
initialized.
|
|
|
|
npackages: limit to N packages
|
|
package_list: limit to those packages in this list
|
|
"""
|
|
WALK_USEFUL_SUBDIRS = ["boot", "linux", "package", "toolchain"]
|
|
WALK_EXCLUDES = ["boot/common.mk",
|
|
"linux/linux-ext-.*.mk",
|
|
"package/freescale-imx/freescale-imx.mk",
|
|
"package/gcc/gcc.mk",
|
|
"package/gstreamer/gstreamer.mk",
|
|
"package/gstreamer1/gstreamer1.mk",
|
|
"package/gtk2-themes/gtk2-themes.mk",
|
|
"package/matchbox/matchbox.mk",
|
|
"package/opengl/opengl.mk",
|
|
"package/qt5/qt5.mk",
|
|
"package/x11r7/x11r7.mk",
|
|
"package/doc-asciidoc.mk",
|
|
"package/pkg-.*.mk",
|
|
"package/nvidia-tegra23/nvidia-tegra23.mk",
|
|
"toolchain/toolchain-external/pkg-toolchain-external.mk",
|
|
"toolchain/toolchain-external/toolchain-external.mk",
|
|
"toolchain/toolchain.mk",
|
|
"toolchain/helpers.mk",
|
|
"toolchain/toolchain-wrapper.mk"]
|
|
packages = list()
|
|
count = 0
|
|
for root, dirs, files in os.walk("."):
|
|
rootdir = root.split("/")
|
|
if len(rootdir) < 2:
|
|
continue
|
|
if rootdir[1] not in WALK_USEFUL_SUBDIRS:
|
|
continue
|
|
for f in files:
|
|
if not f.endswith(".mk"):
|
|
continue
|
|
# Strip ending ".mk"
|
|
pkgname = f[:-3]
|
|
if package_list and pkgname not in package_list:
|
|
continue
|
|
pkgpath = os.path.join(root, f)
|
|
skip = False
|
|
for exclude in WALK_EXCLUDES:
|
|
# pkgpath[2:] strips the initial './'
|
|
if re.match(exclude, pkgpath[2:]):
|
|
skip = True
|
|
continue
|
|
if skip:
|
|
continue
|
|
p = Package(pkgname, pkgpath)
|
|
packages.append(p)
|
|
count += 1
|
|
if npackages and count == npackages:
|
|
return packages
|
|
return packages
|
|
|
|
|
|
def package_init_make_info():
|
|
# Fetch all variables at once
|
|
variables = subprocess.check_output(["make", "BR2_HAVE_DOT_CONFIG=y", "-s", "printvars",
|
|
"VARS=%_LICENSE %_LICENSE_FILES %_VERSION %_IGNORE_CVES"])
|
|
variable_list = variables.splitlines()
|
|
|
|
# We process first the host package VERSION, and then the target
|
|
# package VERSION. This means that if a package exists in both
|
|
# target and host variants, with different values (eg. version
|
|
# numbers (unlikely)), we'll report the target one.
|
|
variable_list = [x[5:] for x in variable_list if x.startswith("HOST_")] + \
|
|
[x for x in variable_list if not x.startswith("HOST_")]
|
|
|
|
for l in variable_list:
|
|
# Get variable name and value
|
|
pkgvar, value = l.split("=")
|
|
|
|
# Strip the suffix according to the variable
|
|
if pkgvar.endswith("_LICENSE"):
|
|
# If value is "unknown", no license details available
|
|
if value == "unknown":
|
|
continue
|
|
pkgvar = pkgvar[:-8]
|
|
Package.all_licenses.append(pkgvar)
|
|
|
|
elif pkgvar.endswith("_LICENSE_FILES"):
|
|
if pkgvar.endswith("_MANIFEST_LICENSE_FILES"):
|
|
continue
|
|
pkgvar = pkgvar[:-14]
|
|
Package.all_license_files.append(pkgvar)
|
|
|
|
elif pkgvar.endswith("_VERSION"):
|
|
if pkgvar.endswith("_DL_VERSION"):
|
|
continue
|
|
pkgvar = pkgvar[:-8]
|
|
Package.all_versions[pkgvar] = value
|
|
|
|
elif pkgvar.endswith("_IGNORE_CVES"):
|
|
pkgvar = pkgvar[:-12]
|
|
Package.all_ignored_cves[pkgvar] = value.split()
|
|
|
|
|
|
def check_url_status_worker(url, url_status):
|
|
if url_status != "Missing" and url_status != "No Config.in":
|
|
try:
|
|
url_status_code = requests.head(url, timeout=30).status_code
|
|
if url_status_code >= 400:
|
|
return "Invalid(%s)" % str(url_status_code)
|
|
except requests.exceptions.RequestException:
|
|
return "Invalid(Err)"
|
|
return "Ok"
|
|
return url_status
|
|
|
|
|
|
def check_package_urls(packages):
|
|
Package.pool = Pool(processes=64)
|
|
for pkg in packages:
|
|
pkg.url_worker = pkg.pool.apply_async(check_url_status_worker, (pkg.url, pkg.url_status))
|
|
for pkg in packages:
|
|
pkg.url_status = pkg.url_worker.get(timeout=3600)
|
|
|
|
|
|
def release_monitoring_get_latest_version_by_distro(pool, name):
|
|
try:
|
|
req = pool.request('GET', "/api/project/Buildroot/%s" % name)
|
|
except HTTPError:
|
|
return (RM_API_STATUS_ERROR, None, None)
|
|
|
|
if req.status != 200:
|
|
return (RM_API_STATUS_NOT_FOUND, None, None)
|
|
|
|
data = json.loads(req.data)
|
|
|
|
if 'version' in data:
|
|
return (RM_API_STATUS_FOUND_BY_DISTRO, data['version'], data['id'])
|
|
else:
|
|
return (RM_API_STATUS_FOUND_BY_DISTRO, None, data['id'])
|
|
|
|
|
|
def release_monitoring_get_latest_version_by_guess(pool, name):
|
|
try:
|
|
req = pool.request('GET', "/api/projects/?pattern=%s" % name)
|
|
except HTTPError:
|
|
return (RM_API_STATUS_ERROR, None, None)
|
|
|
|
if req.status != 200:
|
|
return (RM_API_STATUS_NOT_FOUND, None, None)
|
|
|
|
data = json.loads(req.data)
|
|
|
|
projects = data['projects']
|
|
projects.sort(key=lambda x: x['id'])
|
|
|
|
for p in projects:
|
|
if p['name'] == name and 'version' in p:
|
|
return (RM_API_STATUS_FOUND_BY_PATTERN, p['version'], p['id'])
|
|
|
|
return (RM_API_STATUS_NOT_FOUND, None, None)
|
|
|
|
|
|
def check_package_latest_version_worker(name):
|
|
"""Wrapper to try both by name then by guess"""
|
|
print(name)
|
|
res = release_monitoring_get_latest_version_by_distro(http_pool, name)
|
|
if res[0] == RM_API_STATUS_NOT_FOUND:
|
|
res = release_monitoring_get_latest_version_by_guess(http_pool, name)
|
|
return res
|
|
|
|
|
|
def check_package_latest_version(packages):
|
|
"""
|
|
Fills in the .latest_version field of all Package objects
|
|
|
|
This field has a special format:
|
|
(status, version, id)
|
|
with:
|
|
- status: one of RM_API_STATUS_ERROR,
|
|
RM_API_STATUS_FOUND_BY_DISTRO, RM_API_STATUS_FOUND_BY_PATTERN,
|
|
RM_API_STATUS_NOT_FOUND
|
|
- version: string containing the latest version known by
|
|
release-monitoring.org for this package
|
|
- id: string containing the id of the project corresponding to this
|
|
package, as known by release-monitoring.org
|
|
"""
|
|
global http_pool
|
|
http_pool = HTTPSConnectionPool('release-monitoring.org', port=443,
|
|
cert_reqs='CERT_REQUIRED', ca_certs=certifi.where(),
|
|
timeout=30)
|
|
worker_pool = Pool(processes=64)
|
|
results = worker_pool.map(check_package_latest_version_worker, (pkg.name for pkg in packages))
|
|
for pkg, r in zip(packages, results):
|
|
pkg.latest_version = r
|
|
del http_pool
|
|
|
|
|
|
def check_package_cves(nvd_path, packages):
|
|
if not os.path.isdir(nvd_path):
|
|
os.makedirs(nvd_path)
|
|
|
|
for cve in CVE.read_nvd_dir(nvd_path):
|
|
for pkg_name in cve.pkg_names:
|
|
if pkg_name in packages and cve.affects(packages[pkg_name]):
|
|
packages[pkg_name].cves.append(cve.identifier)
|
|
|
|
|
|
def calculate_stats(packages):
|
|
stats = defaultdict(int)
|
|
for pkg in packages:
|
|
# If packages have multiple infra, take the first one. For the
|
|
# vast majority of packages, the target and host infra are the
|
|
# same. There are very few packages that use a different infra
|
|
# for the host and target variants.
|
|
if len(pkg.infras) > 0:
|
|
infra = pkg.infras[0][1]
|
|
stats["infra-%s" % infra] += 1
|
|
else:
|
|
stats["infra-unknown"] += 1
|
|
if pkg.has_license:
|
|
stats["license"] += 1
|
|
else:
|
|
stats["no-license"] += 1
|
|
if pkg.has_license_files:
|
|
stats["license-files"] += 1
|
|
else:
|
|
stats["no-license-files"] += 1
|
|
if pkg.has_hash:
|
|
stats["hash"] += 1
|
|
else:
|
|
stats["no-hash"] += 1
|
|
if pkg.latest_version[0] == RM_API_STATUS_FOUND_BY_DISTRO:
|
|
stats["rmo-mapping"] += 1
|
|
else:
|
|
stats["rmo-no-mapping"] += 1
|
|
if not pkg.latest_version[1]:
|
|
stats["version-unknown"] += 1
|
|
elif pkg.latest_version[1] == pkg.current_version:
|
|
stats["version-uptodate"] += 1
|
|
else:
|
|
stats["version-not-uptodate"] += 1
|
|
stats["patches"] += pkg.patch_count
|
|
stats["total-cves"] += len(pkg.cves)
|
|
if len(pkg.cves) != 0:
|
|
stats["pkg-cves"] += 1
|
|
return stats
|
|
|
|
|
|
html_header = """
|
|
<head>
|
|
<script src=\"https://www.kryogenix.org/code/browser/sorttable/sorttable.js\"></script>
|
|
<style type=\"text/css\">
|
|
table {
|
|
width: 100%;
|
|
}
|
|
td {
|
|
border: 1px solid black;
|
|
}
|
|
td.centered {
|
|
text-align: center;
|
|
}
|
|
td.wrong {
|
|
background: #ff9a69;
|
|
}
|
|
td.correct {
|
|
background: #d2ffc4;
|
|
}
|
|
td.nopatches {
|
|
background: #d2ffc4;
|
|
}
|
|
td.somepatches {
|
|
background: #ffd870;
|
|
}
|
|
td.lotsofpatches {
|
|
background: #ff9a69;
|
|
}
|
|
|
|
td.good_url {
|
|
background: #d2ffc4;
|
|
}
|
|
td.missing_url {
|
|
background: #ffd870;
|
|
}
|
|
td.invalid_url {
|
|
background: #ff9a69;
|
|
}
|
|
|
|
td.version-good {
|
|
background: #d2ffc4;
|
|
}
|
|
td.version-needs-update {
|
|
background: #ff9a69;
|
|
}
|
|
td.version-unknown {
|
|
background: #ffd870;
|
|
}
|
|
td.version-error {
|
|
background: #ccc;
|
|
}
|
|
|
|
</style>
|
|
<title>Statistics of Buildroot packages</title>
|
|
</head>
|
|
|
|
<a href=\"#results\">Results</a><br/>
|
|
|
|
<p id=\"sortable_hint\"></p>
|
|
"""
|
|
|
|
|
|
html_footer = """
|
|
</body>
|
|
<script>
|
|
if (typeof sorttable === \"object\") {
|
|
document.getElementById(\"sortable_hint\").innerHTML =
|
|
\"hint: the table can be sorted by clicking the column headers\"
|
|
}
|
|
</script>
|
|
</html>
|
|
"""
|
|
|
|
|
|
def infra_str(infra_list):
|
|
if not infra_list:
|
|
return "Unknown"
|
|
elif len(infra_list) == 1:
|
|
return "<b>%s</b><br/>%s" % (infra_list[0][1], infra_list[0][0])
|
|
elif infra_list[0][1] == infra_list[1][1]:
|
|
return "<b>%s</b><br/>%s + %s" % \
|
|
(infra_list[0][1], infra_list[0][0], infra_list[1][0])
|
|
else:
|
|
return "<b>%s</b> (%s)<br/><b>%s</b> (%s)" % \
|
|
(infra_list[0][1], infra_list[0][0],
|
|
infra_list[1][1], infra_list[1][0])
|
|
|
|
|
|
def boolean_str(b):
|
|
if b:
|
|
return "Yes"
|
|
else:
|
|
return "No"
|
|
|
|
|
|
def dump_html_pkg(f, pkg):
|
|
f.write(" <tr>\n")
|
|
f.write(" <td>%s</td>\n" % pkg.path[2:])
|
|
|
|
# Patch count
|
|
td_class = ["centered"]
|
|
if pkg.patch_count == 0:
|
|
td_class.append("nopatches")
|
|
elif pkg.patch_count < 5:
|
|
td_class.append("somepatches")
|
|
else:
|
|
td_class.append("lotsofpatches")
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), str(pkg.patch_count)))
|
|
|
|
# Infrastructure
|
|
infra = infra_str(pkg.infras)
|
|
td_class = ["centered"]
|
|
if infra == "Unknown":
|
|
td_class.append("wrong")
|
|
else:
|
|
td_class.append("correct")
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), infra_str(pkg.infras)))
|
|
|
|
# License
|
|
td_class = ["centered"]
|
|
if pkg.has_license:
|
|
td_class.append("correct")
|
|
else:
|
|
td_class.append("wrong")
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), boolean_str(pkg.has_license)))
|
|
|
|
# License files
|
|
td_class = ["centered"]
|
|
if pkg.has_license_files:
|
|
td_class.append("correct")
|
|
else:
|
|
td_class.append("wrong")
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), boolean_str(pkg.has_license_files)))
|
|
|
|
# Hash
|
|
td_class = ["centered"]
|
|
if pkg.has_hash:
|
|
td_class.append("correct")
|
|
else:
|
|
td_class.append("wrong")
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), boolean_str(pkg.has_hash)))
|
|
|
|
# Current version
|
|
if len(pkg.current_version) > 20:
|
|
current_version = pkg.current_version[:20] + "..."
|
|
else:
|
|
current_version = pkg.current_version
|
|
f.write(" <td class=\"centered\">%s</td>\n" % current_version)
|
|
|
|
# Latest version
|
|
if pkg.latest_version[0] == RM_API_STATUS_ERROR:
|
|
td_class.append("version-error")
|
|
if pkg.latest_version[1] is None:
|
|
td_class.append("version-unknown")
|
|
elif pkg.latest_version[1] != pkg.current_version:
|
|
td_class.append("version-needs-update")
|
|
else:
|
|
td_class.append("version-good")
|
|
|
|
if pkg.latest_version[0] == RM_API_STATUS_ERROR:
|
|
latest_version_text = "<b>Error</b>"
|
|
elif pkg.latest_version[0] == RM_API_STATUS_NOT_FOUND:
|
|
latest_version_text = "<b>Not found</b>"
|
|
else:
|
|
if pkg.latest_version[1] is None:
|
|
latest_version_text = "<b>Found, but no version</b>"
|
|
else:
|
|
latest_version_text = "<a href=\"https://release-monitoring.org/project/%s\"><b>%s</b></a>" % \
|
|
(pkg.latest_version[2], str(pkg.latest_version[1]))
|
|
|
|
latest_version_text += "<br/>"
|
|
|
|
if pkg.latest_version[0] == RM_API_STATUS_FOUND_BY_DISTRO:
|
|
latest_version_text += "found by <a href=\"https://release-monitoring.org/distro/Buildroot/\">distro</a>"
|
|
else:
|
|
latest_version_text += "found by guess"
|
|
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), latest_version_text))
|
|
|
|
# Warnings
|
|
td_class = ["centered"]
|
|
if pkg.warnings == 0:
|
|
td_class.append("correct")
|
|
else:
|
|
td_class.append("wrong")
|
|
f.write(" <td class=\"%s\">%d</td>\n" %
|
|
(" ".join(td_class), pkg.warnings))
|
|
|
|
# URL status
|
|
td_class = ["centered"]
|
|
url_str = pkg.url_status
|
|
if pkg.url_status == "Missing" or pkg.url_status == "No Config.in":
|
|
td_class.append("missing_url")
|
|
elif pkg.url_status.startswith("Invalid"):
|
|
td_class.append("invalid_url")
|
|
url_str = "<a href=%s>%s</a>" % (pkg.url, pkg.url_status)
|
|
else:
|
|
td_class.append("good_url")
|
|
url_str = "<a href=%s>Link</a>" % pkg.url
|
|
f.write(" <td class=\"%s\">%s</td>\n" %
|
|
(" ".join(td_class), url_str))
|
|
|
|
# CVEs
|
|
td_class = ["centered"]
|
|
if len(pkg.cves) == 0:
|
|
td_class.append("correct")
|
|
else:
|
|
td_class.append("wrong")
|
|
f.write(" <td class=\"%s\">\n" % " ".join(td_class))
|
|
for cve in pkg.cves:
|
|
f.write(" <a href=\"https://security-tracker.debian.org/tracker/%s\">%s<br/>\n" % (cve, cve))
|
|
f.write(" </td>\n")
|
|
|
|
f.write(" </tr>\n")
|
|
|
|
|
|
def dump_html_all_pkgs(f, packages):
|
|
f.write("""
|
|
<table class=\"sortable\">
|
|
<tr>
|
|
<td>Package</td>
|
|
<td class=\"centered\">Patch count</td>
|
|
<td class=\"centered\">Infrastructure</td>
|
|
<td class=\"centered\">License</td>
|
|
<td class=\"centered\">License files</td>
|
|
<td class=\"centered\">Hash file</td>
|
|
<td class=\"centered\">Current version</td>
|
|
<td class=\"centered\">Latest version</td>
|
|
<td class=\"centered\">Warnings</td>
|
|
<td class=\"centered\">Upstream URL</td>
|
|
<td class=\"centered\">CVEs</td>
|
|
</tr>
|
|
""")
|
|
for pkg in sorted(packages):
|
|
dump_html_pkg(f, pkg)
|
|
f.write("</table>")
|
|
|
|
|
|
def dump_html_stats(f, stats):
|
|
f.write("<a id=\"results\"></a>\n")
|
|
f.write("<table>\n")
|
|
infras = [infra[6:] for infra in stats.keys() if infra.startswith("infra-")]
|
|
for infra in infras:
|
|
f.write(" <tr><td>Packages using the <i>%s</i> infrastructure</td><td>%s</td></tr>\n" %
|
|
(infra, stats["infra-%s" % infra]))
|
|
f.write(" <tr><td>Packages having license information</td><td>%s</td></tr>\n" %
|
|
stats["license"])
|
|
f.write(" <tr><td>Packages not having license information</td><td>%s</td></tr>\n" %
|
|
stats["no-license"])
|
|
f.write(" <tr><td>Packages having license files information</td><td>%s</td></tr>\n" %
|
|
stats["license-files"])
|
|
f.write(" <tr><td>Packages not having license files information</td><td>%s</td></tr>\n" %
|
|
stats["no-license-files"])
|
|
f.write(" <tr><td>Packages having a hash file</td><td>%s</td></tr>\n" %
|
|
stats["hash"])
|
|
f.write(" <tr><td>Packages not having a hash file</td><td>%s</td></tr>\n" %
|
|
stats["no-hash"])
|
|
f.write(" <tr><td>Total number of patches</td><td>%s</td></tr>\n" %
|
|
stats["patches"])
|
|
f.write("<tr><td>Packages having a mapping on <i>release-monitoring.org</i></td><td>%s</td></tr>\n" %
|
|
stats["rmo-mapping"])
|
|
f.write("<tr><td>Packages lacking a mapping on <i>release-monitoring.org</i></td><td>%s</td></tr>\n" %
|
|
stats["rmo-no-mapping"])
|
|
f.write("<tr><td>Packages that are up-to-date</td><td>%s</td></tr>\n" %
|
|
stats["version-uptodate"])
|
|
f.write("<tr><td>Packages that are not up-to-date</td><td>%s</td></tr>\n" %
|
|
stats["version-not-uptodate"])
|
|
f.write("<tr><td>Packages with no known upstream version</td><td>%s</td></tr>\n" %
|
|
stats["version-unknown"])
|
|
f.write("<tr><td>Packages affected by CVEs</td><td>%s</td></tr>\n" %
|
|
stats["pkg-cves"])
|
|
f.write("<tr><td>Total number of CVEs affecting all packages</td><td>%s</td></tr>\n" %
|
|
stats["total-cves"])
|
|
f.write("</table>\n")
|
|
|
|
|
|
def dump_html_gen_info(f, date, commit):
|
|
# Updated on Mon Feb 19 08:12:08 CET 2018, Git commit aa77030b8f5e41f1c53eb1c1ad664b8c814ba032
|
|
f.write("<p><i>Updated on %s, git commit %s</i></p>\n" % (str(date), commit))
|
|
|
|
|
|
def dump_html(packages, stats, date, commit, output):
|
|
with open(output, 'w') as f:
|
|
f.write(html_header)
|
|
dump_html_all_pkgs(f, packages)
|
|
dump_html_stats(f, stats)
|
|
dump_html_gen_info(f, date, commit)
|
|
f.write(html_footer)
|
|
|
|
|
|
def dump_json(packages, stats, date, commit, output):
|
|
# Format packages as a dictionnary instead of a list
|
|
# Exclude local field that does not contains real date
|
|
excluded_fields = ['url_worker', 'name']
|
|
pkgs = {
|
|
pkg.name: {
|
|
k: v
|
|
for k, v in pkg.__dict__.items()
|
|
if k not in excluded_fields
|
|
} for pkg in packages
|
|
}
|
|
# Aggregate infrastructures into a single dict entry
|
|
statistics = {
|
|
k: v
|
|
for k, v in stats.items()
|
|
if not k.startswith('infra-')
|
|
}
|
|
statistics['infra'] = {k[6:]: v for k, v in stats.items() if k.startswith('infra-')}
|
|
# The actual structure to dump, add commit and date to it
|
|
final = {'packages': pkgs,
|
|
'stats': statistics,
|
|
'commit': commit,
|
|
'date': str(date)}
|
|
|
|
with open(output, 'w') as f:
|
|
json.dump(final, f, indent=2, separators=(',', ': '))
|
|
f.write('\n')
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser()
|
|
output = parser.add_argument_group('output', 'Output file(s)')
|
|
output.add_argument('--html', dest='html', action='store',
|
|
help='HTML output file')
|
|
output.add_argument('--json', dest='json', action='store',
|
|
help='JSON output file')
|
|
packages = parser.add_mutually_exclusive_group()
|
|
packages.add_argument('-n', dest='npackages', type=int, action='store',
|
|
help='Number of packages')
|
|
packages.add_argument('-p', dest='packages', action='store',
|
|
help='List of packages (comma separated)')
|
|
parser.add_argument('--nvd-path', dest='nvd_path',
|
|
help='Path to the local NVD database')
|
|
args = parser.parse_args()
|
|
if not args.html and not args.json:
|
|
parser.error('at least one of --html or --json (or both) is required')
|
|
return args
|
|
|
|
|
|
def __main__():
|
|
args = parse_args()
|
|
if args.packages:
|
|
package_list = args.packages.split(",")
|
|
else:
|
|
package_list = None
|
|
date = datetime.datetime.utcnow()
|
|
commit = subprocess.check_output(['git', 'rev-parse',
|
|
'HEAD']).splitlines()[0]
|
|
print("Build package list ...")
|
|
packages = get_pkglist(args.npackages, package_list)
|
|
print("Getting package make info ...")
|
|
package_init_make_info()
|
|
print("Getting package details ...")
|
|
for pkg in packages:
|
|
pkg.set_infra()
|
|
pkg.set_license()
|
|
pkg.set_hash_info()
|
|
pkg.set_patch_count()
|
|
pkg.set_check_package_warnings()
|
|
pkg.set_current_version()
|
|
pkg.set_url()
|
|
print("Checking URL status")
|
|
check_package_urls(packages)
|
|
print("Getting latest versions ...")
|
|
check_package_latest_version(packages)
|
|
if args.nvd_path:
|
|
print("Checking packages CVEs")
|
|
check_package_cves(args.nvd_path, {p.name: p for p in packages})
|
|
print("Calculate stats")
|
|
stats = calculate_stats(packages)
|
|
if args.html:
|
|
print("Write HTML")
|
|
dump_html(packages, stats, date, commit, args.html)
|
|
if args.json:
|
|
print("Write JSON")
|
|
dump_json(packages, stats, date, commit, args.json)
|
|
|
|
|
|
__main__()
|