1 files changed, 195 insertions, 62 deletions

diff --git a/meta/lib/oe/cve_check.py b/meta/lib/oe/cve_check.py
index ed5c714cb8..ae194f27cf 100644
--- a/meta/lib/oe/cve_check.py
+++ b/meta/lib/oe/cve_check.py
@@ -5,9 +5,11 @@
 #
 
 import collections
-import re
-import itertools
 import functools
+import itertools
+import os.path
+import re
+import oe.patch
 
 _Version = collections.namedtuple(
     "_Version", ["release", "patch_l", "pre_l", "pre_v"]
@@ -71,71 +73,134 @@ def _cmpkey(release, patch_l, pre_l, pre_v):
     return _release, _patch, _pre
 
 
-def get_patched_cves(d):
+def parse_cve_from_filename(patch_filename):
     """
-    Get patches that solve CVEs using the "CVE: " tag.
+    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)
 
-    import re
-    import oe.patch
+    # 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 ""
 
-    cve_match = re.compile(r"CVE:( CVE-\d{4}-\d+)+")
 
-    # 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.
-    # However, patch files contents addressing multiple CVE IDs are supported
-    # (cve_match regular expression)
-    cve_file_name_match = re.compile(r".*(CVE-\d{4}-\d+)", re.IGNORECASE)
+def parse_cves_from_patch_contents(patch_contents):
+    """
+    Parses CVE IDs from patch contents
 
-    patched_cves = set()
-    patches = oe.patch.src_patches(d)
-    bb.debug(2, "Scanning %d patches for CVEs" % len(patches))
-    for url in patches:
-        patch_file = bb.fetch.decodeurl(url)[2]
+    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.
 
-        # Check patch file name for CVE ID
-        fname_match = cve_file_name_match.search(patch_file)
-        if fname_match:
-            cve = fname_match.group(1).upper()
-            patched_cves.add(cve)
-            bb.debug(2, "Found %s from patch file name %s" % (cve, 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)
+    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
+
 
-        with open(patch_file, "r", encoding="utf-8") as f:
-            try:
+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()
-            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()
-
-        # Search for one or more "CVE: " lines
-        text_match = False
-        for match in cve_match.finditer(patch_text):
-            # Get only the CVEs without the "CVE: " tag
-            cves = patch_text[match.start()+5:match.end()]
-            for cve in cves.split():
-                bb.debug(2, "Patch %s solves %s" % (patch_file, cve))
-                patched_cves.add(cve)
-                text_match = True
-
-        if not fname_match and not text_match:
-            bb.debug(2, "Patch %s doesn't solve CVEs" % patch_file)
+
+    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
+
+
+@bb.parse.vardeps("CVE_STATUS")
+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 in (d.getVarFlags("CVE_STATUS") or {}):
-        decoded_status, _, _ = decode_cve_status(d, cve)
-        if decoded_status == "Patched":
-            bb.debug(2, "CVE %s is additionally patched" % cve)
-            patched_cves.add(cve)
+    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
 
@@ -225,21 +290,89 @@ def convert_cve_version(version):
 
     return version + update
 
+@bb.parse.vardeps("CVE_STATUS", "CVE_CHECK_STATUSMAP")
 def decode_cve_status(d, cve):
     """
-    Convert CVE_STATUS into status, detail and description.
+    Convert CVE_STATUS into status, vendor, product, detail and description.
     """
     status = d.getVarFlag("CVE_STATUS", cve)
     if not status:
-        return ("", "", "")
+        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_split = status.split(':', 1)
-    detail = status_split[0]
-    description = status_split[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))
+        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
 
-    return (status_mapping, detail, description)
+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
+
+def extend_cve_status(d):
+    # do this only once in case multiple classes use this
+    if d.getVar("CVE_STATUS_EXTENDED"):
+        return
+    d.setVar("CVE_STATUS_EXTENDED", "1")
+
+    # Fallback all CVEs from CVE_CHECK_IGNORE to CVE_STATUS
+    cve_check_ignore = d.getVar("CVE_CHECK_IGNORE")
+    if cve_check_ignore:
+        bb.warn("CVE_CHECK_IGNORE is deprecated in favor of CVE_STATUS")
+        for cve in (d.getVar("CVE_CHECK_IGNORE") or "").split():
+            d.setVarFlag("CVE_STATUS", cve, "ignored")
+
+    # Process CVE_STATUS_GROUPS to set multiple statuses and optional detail or description at once
+    for cve_status_group in (d.getVar("CVE_STATUS_GROUPS") or "").split():
+        cve_group = d.getVar(cve_status_group)
+        if cve_group is not None:
+            for cve in cve_group.split():
+                d.setVarFlag("CVE_STATUS", cve, d.getVarFlag(cve_status_group, "status"))
+        else:
+            bb.warn("CVE_STATUS_GROUPS contains undefined variable %s" % cve_status_group)