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