xref: /openbmc/openbmc/poky/meta/lib/oe/cve_check.py (revision 92b42cb3)
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    pn = d.getVar("PN")
83    cve_match = re.compile("CVE:( CVE\-\d{4}\-\d+)+")
84
85    # Matches the last "CVE-YYYY-ID" in the file name, also if written
86    # in lowercase. Possible to have multiple CVE IDs in a single
87    # file name, but only the last one will be detected from the file name.
88    # However, patch files contents addressing multiple CVE IDs are supported
89    # (cve_match regular expression)
90
91    cve_file_name_match = re.compile(".*([Cc][Vv][Ee]\-\d{4}\-\d+)")
92
93    patched_cves = set()
94    bb.debug(2, "Looking for patches that solves CVEs for %s" % pn)
95    for url in oe.patch.src_patches(d):
96        patch_file = bb.fetch.decodeurl(url)[2]
97
98        # Remote compressed patches may not be unpacked, so silently ignore them
99        if not os.path.isfile(patch_file):
100            bb.warn("%s does not exist, cannot extract CVE list" % patch_file)
101            continue
102
103        # Check patch file name for CVE ID
104        fname_match = cve_file_name_match.search(patch_file)
105        if fname_match:
106            cve = fname_match.group(1).upper()
107            patched_cves.add(cve)
108            bb.debug(2, "Found CVE %s from patch file name %s" % (cve, patch_file))
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.add(cve)
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    return patched_cves
134
135
136def get_cpe_ids(cve_product, version):
137    """
138    Get list of CPE identifiers for the given product and version
139    """
140
141    version = version.split("+git")[0]
142
143    cpe_ids = []
144    for product in cve_product.split():
145        # CVE_PRODUCT in recipes may include vendor information for CPE identifiers. If not,
146        # use wildcard for vendor.
147        if ":" in product:
148            vendor, product = product.split(":", 1)
149        else:
150            vendor = "*"
151
152        cpe_id = 'cpe:2.3:a:{}:{}:{}:*:*:*:*:*:*:*'.format(vendor, product, version)
153        cpe_ids.append(cpe_id)
154
155    return cpe_ids
156
157def cve_check_merge_jsons(output, data):
158    """
159    Merge the data in the "package" property to the main data file
160    output
161    """
162    if output["version"] != data["version"]:
163        bb.error("Version mismatch when merging JSON outputs")
164        return
165
166    for product in output["package"]:
167        if product["name"] == data["package"][0]["name"]:
168            bb.error("Error adding the same package twice")
169            return
170
171    output["package"].append(data["package"][0])
172
173def update_symlinks(target_path, link_path):
174    """
175    Update a symbolic link link_path to point to target_path.
176    Remove the link and recreate it if exist and is different.
177    """
178    if link_path != target_path and os.path.exists(target_path):
179        if os.path.exists(os.path.realpath(link_path)):
180            os.remove(link_path)
181        os.symlink(os.path.basename(target_path), link_path)
182