xref: /openbmc/openbmc/poky/meta/lib/oe/cve_check.py (revision 8460358c)
1#
2# Copyright OpenEmbedded Contributors
3#
4# SPDX-License-Identifier: MIT
5#
6
7import collections
8import re
9import itertools
10import functools
11
12_Version = collections.namedtuple(
13    "_Version", ["release", "patch_l", "pre_l", "pre_v"]
14)
15
16@functools.total_ordering
17class Version():
18
19    def __init__(self, version, suffix=None):
20
21        suffixes = ["alphabetical", "patch"]
22
23        if str(suffix) == "alphabetical":
24            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]+)?)?)(.*)?"""
25        elif str(suffix) == "patch":
26            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]+)?)?)(.*)?"""
27        else:
28            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]+)?)?)(.*)?"""
29        regex = re.compile(r"^\s*" + version_pattern + r"\s*$", re.VERBOSE | re.IGNORECASE)
30
31        match = regex.search(version)
32        if not match:
33            raise Exception("Invalid version: '{0}'".format(version))
34
35        self._version = _Version(
36            release=tuple(int(i) for i in match.group("release").replace("-",".").split(".")),
37            patch_l=match.group("patch_l") if str(suffix) in suffixes and match.group("patch_l") else "",
38            pre_l=match.group("pre_l"),
39            pre_v=match.group("pre_v")
40        )
41
42        self._key = _cmpkey(
43            self._version.release,
44            self._version.patch_l,
45            self._version.pre_l,
46            self._version.pre_v
47        )
48
49    def __eq__(self, other):
50        if not isinstance(other, Version):
51            return NotImplemented
52        return self._key == other._key
53
54    def __gt__(self, other):
55        if not isinstance(other, Version):
56            return NotImplemented
57        return self._key > other._key
58
59def _cmpkey(release, patch_l, pre_l, pre_v):
60    # remove leading 0
61    _release = tuple(
62        reversed(list(itertools.dropwhile(lambda x: x == 0, reversed(release))))
63    )
64
65    _patch = patch_l.upper()
66
67    if pre_l is None and pre_v is None:
68        _pre = float('inf')
69    else:
70        _pre = float(pre_v) if pre_v else float('-inf')
71    return _release, _patch, _pre
72
73
74def get_patched_cves(d):
75    """
76    Get patches that solve CVEs using the "CVE: " tag.
77    """
78
79    import re
80    import oe.patch
81
82    cve_match = re.compile(r"CVE:( CVE-\d{4}-\d+)+")
83
84    # Matches the last "CVE-YYYY-ID" in the file name, also if written
85    # in lowercase. Possible to have multiple CVE IDs in a single
86    # file name, but only the last one will be detected from the file name.
87    # However, patch files contents addressing multiple CVE IDs are supported
88    # (cve_match regular expression)
89    cve_file_name_match = re.compile(r".*(CVE-\d{4}-\d+)", re.IGNORECASE)
90
91    patched_cves = {}
92    patches = oe.patch.src_patches(d)
93    bb.debug(2, "Scanning %d patches for CVEs" % len(patches))
94    for url in patches:
95        patch_file = bb.fetch.decodeurl(url)[2]
96
97        # Check patch file name for CVE ID
98        fname_match = cve_file_name_match.search(patch_file)
99        if fname_match:
100            cve = fname_match.group(1).upper()
101            patched_cves[cve] = {"abbrev-status": "Patched", "status": "fix-file-included", "resource": patch_file}
102            bb.debug(2, "Found %s from patch file name %s" % (cve, patch_file))
103
104        # Remote patches won't be present and compressed patches won't be
105        # unpacked, so say we're not scanning them
106        if not os.path.isfile(patch_file):
107            bb.note("%s is remote or compressed, not scanning content" % patch_file)
108            continue
109
110        with open(patch_file, "r", encoding="utf-8") as f:
111            try:
112                patch_text = f.read()
113            except UnicodeDecodeError:
114                bb.debug(1, "Failed to read patch %s using UTF-8 encoding"
115                        " trying with iso8859-1" %  patch_file)
116                f.close()
117                with open(patch_file, "r", encoding="iso8859-1") as f:
118                    patch_text = f.read()
119
120        # Search for one or more "CVE: " lines
121        text_match = False
122        for match in cve_match.finditer(patch_text):
123            # Get only the CVEs without the "CVE: " tag
124            cves = patch_text[match.start()+5:match.end()]
125            for cve in cves.split():
126                bb.debug(2, "Patch %s solves %s" % (patch_file, cve))
127                patched_cves[cve] = {"abbrev-status": "Patched", "status": "fix-file-included", "resource": patch_file}
128                text_match = True
129
130        if not fname_match and not text_match:
131            bb.debug(2, "Patch %s doesn't solve CVEs" % patch_file)
132
133    # Search for additional patched CVEs
134    for cve in (d.getVarFlags("CVE_STATUS") or {}):
135        decoded_status = decode_cve_status(d, cve)
136        products = d.getVar("CVE_PRODUCT")
137        if has_cve_product_match(decoded_status, products) == True:
138            patched_cves[cve] = {
139                "abbrev-status": decoded_status["mapping"],
140                "status": decoded_status["detail"],
141                "justification": decoded_status["description"],
142                "affected-vendor": decoded_status["vendor"],
143                "affected-product": decoded_status["product"]
144            }
145
146    return patched_cves
147
148
149def get_cpe_ids(cve_product, version):
150    """
151    Get list of CPE identifiers for the given product and version
152    """
153
154    version = version.split("+git")[0]
155
156    cpe_ids = []
157    for product in cve_product.split():
158        # CVE_PRODUCT in recipes may include vendor information for CPE identifiers. If not,
159        # use wildcard for vendor.
160        if ":" in product:
161            vendor, product = product.split(":", 1)
162        else:
163            vendor = "*"
164
165        cpe_id = 'cpe:2.3:*:{}:{}:{}:*:*:*:*:*:*:*'.format(vendor, product, version)
166        cpe_ids.append(cpe_id)
167
168    return cpe_ids
169
170def cve_check_merge_jsons(output, data):
171    """
172    Merge the data in the "package" property to the main data file
173    output
174    """
175    if output["version"] != data["version"]:
176        bb.error("Version mismatch when merging JSON outputs")
177        return
178
179    for product in output["package"]:
180        if product["name"] == data["package"][0]["name"]:
181            bb.error("Error adding the same package %s twice" % product["name"])
182            return
183
184    output["package"].append(data["package"][0])
185
186def update_symlinks(target_path, link_path):
187    """
188    Update a symbolic link link_path to point to target_path.
189    Remove the link and recreate it if exist and is different.
190    """
191    if link_path != target_path and os.path.exists(target_path):
192        if os.path.exists(os.path.realpath(link_path)):
193            os.remove(link_path)
194        os.symlink(os.path.basename(target_path), link_path)
195
196
197def convert_cve_version(version):
198    """
199    This function converts from CVE format to Yocto version format.
200    eg 8.3_p1 -> 8.3p1, 6.2_rc1 -> 6.2-rc1
201
202    Unless it is redefined using CVE_VERSION in the recipe,
203    cve_check uses the version in the name of the recipe (${PV})
204    to check vulnerabilities against a CVE in the database downloaded from NVD.
205
206    When the version has an update, i.e.
207    "p1" in OpenSSH 8.3p1,
208    "-rc1" in linux kernel 6.2-rc1,
209    the database stores the version as version_update (8.3_p1, 6.2_rc1).
210    Therefore, we must transform this version before comparing to the
211    recipe version.
212
213    In this case, the parameter of the function is 8.3_p1.
214    If the version uses the Release Candidate format, "rc",
215    this function replaces the '_' by '-'.
216    If the version uses the Update format, "p",
217    this function removes the '_' completely.
218    """
219    import re
220
221    matches = re.match('^([0-9.]+)_((p|rc)[0-9]+)$', version)
222
223    if not matches:
224        return version
225
226    version = matches.group(1)
227    update = matches.group(2)
228
229    if matches.group(3) == "rc":
230        return version + '-' + update
231
232    return version + update
233
234def decode_cve_status(d, cve):
235    """
236    Convert CVE_STATUS into status, vendor, product, detail and description.
237    """
238    status = d.getVarFlag("CVE_STATUS", cve)
239    if not status:
240        return {}
241
242    status_split = status.split(':', 4)
243    status_out = {}
244    status_out["detail"] = status_split[0]
245    product = "*"
246    vendor = "*"
247    description = ""
248    if len(status_split) >= 4 and status_split[1].strip() == "cpe":
249        # Both vendor and product are mandatory if cpe: present, the syntax is then:
250        # detail: cpe:vendor:product:description
251        vendor = status_split[2].strip()
252        product = status_split[3].strip()
253        description = status_split[4].strip()
254    elif len(status_split) >= 2 and status_split[1].strip() == "cpe":
255        # Malformed CPE
256        bb.warn('Invalid CPE information for CVE_STATUS[%s] = "%s", not setting CPE' % (detail, cve, status))
257    else:
258        # Other case: no CPE, the syntax is then:
259        # detail: description
260        description = status.split(':', 1)[1].strip() if (len(status_split) > 1) else ""
261
262    status_out["vendor"] = vendor
263    status_out["product"] = product
264    status_out["description"] = description
265
266    status_mapping = d.getVarFlag("CVE_CHECK_STATUSMAP", status_out['detail'])
267    if status_mapping is None:
268        bb.warn('Invalid detail "%s" for CVE_STATUS[%s] = "%s", fallback to Unpatched' % (detail, cve, status))
269        status_mapping = "Unpatched"
270    status_out["mapping"] = status_mapping
271
272    return status_out
273
274def has_cve_product_match(detailed_status, products):
275    """
276    Check product/vendor match between detailed_status from decode_cve_status and a string of
277    products (like from CVE_PRODUCT)
278    """
279    for product in products.split():
280        vendor = "*"
281        if ":" in product:
282            vendor, product = product.split(":", 1)
283
284        if (vendor == detailed_status["vendor"] or detailed_status["vendor"] == "*") and \
285            (product == detailed_status["product"] or detailed_status["product"] == "*"):
286            return True
287
288    #if no match, return False
289    return False
290