blob: 5c272ba4ff34ec7d99c80f30296c23baa775f117 [file] [log] [blame] [edit]
#
# Copyright OpenEmbedded Contributors
#
# SPDX-License-Identifier: MIT
#
import collections
import functools
import itertools
import os.path
import re
import oe.patch
_Version = collections.namedtuple(
"_Version", ["release", "patch_l", "pre_l", "pre_v"]
)
@functools.total_ordering
class Version():
def __init__(self, version, suffix=None):
suffixes = ["alphabetical", "patch"]
if str(suffix) == "alphabetical":
version_pattern = r"""r?v?(?:(?P<release>[0-9]+(?:[-\.][0-9]+)*)(?P<patch>[-_\.]?(?P<patch_l>[a-z]))?(?P<pre>[-_\.]?(?P<pre_l>(rc|alpha|beta|pre|preview|dev))[-_\.]?(?P<pre_v>[0-9]+)?)?)(.*)?"""
elif str(suffix) == "patch":
version_pattern = r"""r?v?(?:(?P<release>[0-9]+(?:[-\.][0-9]+)*)(?P<patch>[-_\.]?(p|patch)(?P<patch_l>[0-9]+))?(?P<pre>[-_\.]?(?P<pre_l>(rc|alpha|beta|pre|preview|dev))[-_\.]?(?P<pre_v>[0-9]+)?)?)(.*)?"""
else:
version_pattern = r"""r?v?(?:(?P<release>[0-9]+(?:[-\.][0-9]+)*)(?P<pre>[-_\.]?(?P<pre_l>(rc|alpha|beta|pre|preview|dev))[-_\.]?(?P<pre_v>[0-9]+)?)?)(.*)?"""
regex = re.compile(r"^\s*" + version_pattern + r"\s*$", re.VERBOSE | re.IGNORECASE)
match = regex.search(version)
if not match:
raise Exception("Invalid version: '{0}'".format(version))
self._version = _Version(
release=tuple(int(i) for i in match.group("release").replace("-",".").split(".")),
patch_l=match.group("patch_l") if str(suffix) in suffixes and match.group("patch_l") else "",
pre_l=match.group("pre_l"),
pre_v=match.group("pre_v")
)
self._key = _cmpkey(
self._version.release,
self._version.patch_l,
self._version.pre_l,
self._version.pre_v
)
def __eq__(self, other):
if not isinstance(other, Version):
return NotImplemented
return self._key == other._key
def __gt__(self, other):
if not isinstance(other, Version):
return NotImplemented
return self._key > other._key
def _cmpkey(release, patch_l, pre_l, pre_v):
# remove leading 0
_release = tuple(
reversed(list(itertools.dropwhile(lambda x: x == 0, reversed(release))))
)
_patch = patch_l.upper()
if pre_l is None and pre_v is None:
_pre = float('inf')
else:
_pre = float(pre_v) if pre_v else float('-inf')
return _release, _patch, _pre
def parse_cve_from_filename(patch_filename):
"""
Parses CVE ID from the filename
Matches the last "CVE-YYYY-ID" in the file name, also if written
in lowercase. Possible to have multiple CVE IDs in a single
file name, but only the last one will be detected from the file name.
Returns the last CVE ID foudn in the filename. If no CVE ID is found
an empty string is returned.
"""
cve_file_name_match = re.compile(r".*(CVE-\d{4}-\d{4,})", re.IGNORECASE)
# Check patch file name for CVE ID
fname_match = cve_file_name_match.search(patch_filename)
return fname_match.group(1).upper() if fname_match else ""
def parse_cves_from_patch_contents(patch_contents):
"""
Parses CVE IDs from patch contents
Matches all CVE IDs contained on a line that starts with "CVE: ". Any
delimiter (',', '&', "and", etc.) can be used without any issues. Multiple
"CVE:" lines can also exist.
Returns a set of all CVE IDs found in the patch contents.
"""
cve_ids = set()
cve_match = re.compile(r"CVE-\d{4}-\d{4,}")
# Search for one or more "CVE: " lines
for line in patch_contents.split("\n"):
if not line.startswith("CVE:"):
continue
cve_ids.update(cve_match.findall(line))
return cve_ids
def parse_cves_from_patch_file(patch_file):
"""
Parses CVE IDs associated with a particular patch file, using both the filename
and patch contents.
Returns a set of all CVE IDs found in the patch filename and contents.
"""
cve_ids = set()
filename_cve = parse_cve_from_filename(patch_file)
if filename_cve:
bb.debug(2, "Found %s from patch file name %s" % (filename_cve, patch_file))
cve_ids.add(parse_cve_from_filename(patch_file))
# Remote patches won't be present and compressed patches won't be
# unpacked, so say we're not scanning them
if not os.path.isfile(patch_file):
bb.note("%s is remote or compressed, not scanning content" % patch_file)
return cve_ids
with open(patch_file, "r", encoding="utf-8") as f:
try:
patch_text = f.read()
except UnicodeDecodeError:
bb.debug(
1,
"Failed to read patch %s using UTF-8 encoding"
" trying with iso8859-1" % patch_file,
)
f.close()
with open(patch_file, "r", encoding="iso8859-1") as f:
patch_text = f.read()
cve_ids.update(parse_cves_from_patch_contents(patch_text))
if not cve_ids:
bb.debug(2, "Patch %s doesn't solve CVEs" % patch_file)
else:
bb.debug(2, "Patch %s solves %s" % (patch_file, ", ".join(sorted(cve_ids))))
return cve_ids
def get_patched_cves(d):
"""
Determines the CVE IDs that have been solved by either patches incuded within
SRC_URI or by setting CVE_STATUS.
Returns a dictionary with the CVE IDs as keys and an associated dictonary of
relevant metadata as the value.
"""
patched_cves = {}
patches = oe.patch.src_patches(d)
bb.debug(2, "Scanning %d patches for CVEs" % len(patches))
# Check each patch file
for url in patches:
patch_file = bb.fetch.decodeurl(url)[2]
for cve_id in parse_cves_from_patch_file(patch_file):
if cve_id not in patched_cves:
patched_cves[cve_id] = {
"abbrev-status": "Patched",
"status": "fix-file-included",
"resource": [patch_file],
}
else:
patched_cves[cve_id]["resource"].append(patch_file)
# Search for additional patched CVEs
for cve_id in d.getVarFlags("CVE_STATUS") or {}:
decoded_status = decode_cve_status(d, cve_id)
products = d.getVar("CVE_PRODUCT")
if has_cve_product_match(decoded_status, products):
if cve_id in patched_cves:
bb.warn(
'CVE_STATUS[%s] = "%s" is overwriting previous status of "%s: %s"'
% (
cve_id,
d.getVarFlag("CVE_STATUS", cve_id),
patched_cves[cve_id]["abbrev-status"],
patched_cves[cve_id]["status"],
)
)
patched_cves[cve_id] = {
"abbrev-status": decoded_status["mapping"],
"status": decoded_status["detail"],
"justification": decoded_status["description"],
"affected-vendor": decoded_status["vendor"],
"affected-product": decoded_status["product"],
}
return patched_cves
def get_cpe_ids(cve_product, version):
"""
Get list of CPE identifiers for the given product and version
"""
version = version.split("+git")[0]
cpe_ids = []
for product in cve_product.split():
# CVE_PRODUCT in recipes may include vendor information for CPE identifiers. If not,
# use wildcard for vendor.
if ":" in product:
vendor, product = product.split(":", 1)
else:
vendor = "*"
cpe_id = 'cpe:2.3:*:{}:{}:{}:*:*:*:*:*:*:*'.format(vendor, product, version)
cpe_ids.append(cpe_id)
return cpe_ids
def cve_check_merge_jsons(output, data):
"""
Merge the data in the "package" property to the main data file
output
"""
if output["version"] != data["version"]:
bb.error("Version mismatch when merging JSON outputs")
return
for product in output["package"]:
if product["name"] == data["package"][0]["name"]:
bb.error("Error adding the same package %s twice" % product["name"])
return
output["package"].append(data["package"][0])
def update_symlinks(target_path, link_path):
"""
Update a symbolic link link_path to point to target_path.
Remove the link and recreate it if exist and is different.
"""
if link_path != target_path and os.path.exists(target_path):
if os.path.exists(os.path.realpath(link_path)):
os.remove(link_path)
os.symlink(os.path.basename(target_path), link_path)
def convert_cve_version(version):
"""
This function converts from CVE format to Yocto version format.
eg 8.3_p1 -> 8.3p1, 6.2_rc1 -> 6.2-rc1
Unless it is redefined using CVE_VERSION in the recipe,
cve_check uses the version in the name of the recipe (${PV})
to check vulnerabilities against a CVE in the database downloaded from NVD.
When the version has an update, i.e.
"p1" in OpenSSH 8.3p1,
"-rc1" in linux kernel 6.2-rc1,
the database stores the version as version_update (8.3_p1, 6.2_rc1).
Therefore, we must transform this version before comparing to the
recipe version.
In this case, the parameter of the function is 8.3_p1.
If the version uses the Release Candidate format, "rc",
this function replaces the '_' by '-'.
If the version uses the Update format, "p",
this function removes the '_' completely.
"""
import re
matches = re.match('^([0-9.]+)_((p|rc)[0-9]+)$', version)
if not matches:
return version
version = matches.group(1)
update = matches.group(2)
if matches.group(3) == "rc":
return version + '-' + update
return version + update
def decode_cve_status(d, cve):
"""
Convert CVE_STATUS into status, vendor, product, detail and description.
"""
status = d.getVarFlag("CVE_STATUS", cve)
if not status:
return {}
status_split = status.split(':', 4)
status_out = {}
status_out["detail"] = status_split[0]
product = "*"
vendor = "*"
description = ""
if len(status_split) >= 4 and status_split[1].strip() == "cpe":
# Both vendor and product are mandatory if cpe: present, the syntax is then:
# detail: cpe:vendor:product:description
vendor = status_split[2].strip()
product = status_split[3].strip()
description = status_split[4].strip()
elif len(status_split) >= 2 and status_split[1].strip() == "cpe":
# Malformed CPE
bb.warn(
'Invalid CPE information for CVE_STATUS[%s] = "%s", not setting CPE'
% (cve, status)
)
else:
# Other case: no CPE, the syntax is then:
# detail: description
description = status.split(':', 1)[1].strip() if (len(status_split) > 1) else ""
status_out["vendor"] = vendor
status_out["product"] = product
status_out["description"] = description
detail = status_out["detail"]
status_mapping = d.getVarFlag("CVE_CHECK_STATUSMAP", detail)
if status_mapping is None:
bb.warn(
'Invalid detail "%s" for CVE_STATUS[%s] = "%s", fallback to Unpatched'
% (detail, cve, status)
)
status_mapping = "Unpatched"
status_out["mapping"] = status_mapping
return status_out
def has_cve_product_match(detailed_status, products):
"""
Check product/vendor match between detailed_status from decode_cve_status and a string of
products (like from CVE_PRODUCT)
"""
for product in products.split():
vendor = "*"
if ":" in product:
vendor, product = product.split(":", 1)
if (vendor == detailed_status["vendor"] or detailed_status["vendor"] == "*") and \
(product == detailed_status["product"] or detailed_status["product"] == "*"):
return True
#if no match, return False
return False