secscan: implement basescore/cveid (PROJQUAY-6697) (#2684)

* secscan: implement basescore/cveid (PROJQUAY-6697)

Signed-off-by: Ross Bryan <robryan@redhat.com>

* docs: add docstrings to helper functions

Signed-off-by: Ross Bryan <robryan@redhat.com>

* secscan: refactor base_score builder loop

Signed-off-by: Ross Bryan <robryan@redhat.com>

* make cveIDs a unique, alpha sorted list

Signed-off-by: Ross Bryan <robryan@redhat.com>

---------

Signed-off-by: Ross Bryan <robryan@redhat.com>

data/secscan_model/datatypes.py

@@ -1,7 +1,47 @@
+import re
 from collections import namedtuple
 from enum import IntEnum, unique
 
 
+def link_to_cves(input_string):
+    """
+    link_to_cves takes an input string, typically the link field from a clair response
+    parses the string and finds all unique CVEs within the string.
+    """
+    cve_pattern = r"CVE-\d{4}-\d{4,7}"
+    return sorted(list(set(re.findall(cve_pattern, input_string))))
+
+
+def vulns_to_cves(vulnerabilities):
+    """
+    vulns_to_cves takes a list of Vulnerabilities and returns
+    a unique list of CVE Ids sorted alphabetically
+    """
+    seen = set()
+    return sorted(
+        [
+            cve
+            for v in vulnerabilities
+            for cve in link_to_cves(v.Link)
+            if not (cve in seen or seen.add(cve))
+        ]
+    )
+
+
+def vulns_to_base_scores(vulnerabilities):
+    """
+    vulns_to_base_scores takes a list of Vulnerabilities and returns a list of
+    CVE BaseScores for the given Vulnerabilities
+    """
+    return [
+        vulnerability.Metadata.NVD.CVSSv3.Score
+        for vulnerability in vulnerabilities
+        if vulnerability.Metadata.NVD
+        and vulnerability.Metadata.NVD.CVSSv3
+        and vulnerability.Metadata.NVD.CVSSv3.Score
+    ]
+
+
 @unique
 class ScanLookupStatus(IntEnum):
     # Indicates that the given manifest or image could not be found in the registry data model.
@@ -39,7 +79,17 @@ class ScanLookupStatus(IntEnum):
 NVD = namedtuple("NVD", ["CVSSv3"])
 CVSSv3 = namedtuple("CVSSv3", ["Vectors", "Score"], defaults=(None, None))
 Feature = namedtuple(
-    "Feature", ["Name", "VersionFormat", "NamespaceName", "AddedBy", "Version", "Vulnerabilities"]
+    "Feature",
+    [
+        "Name",
+        "VersionFormat",
+        "NamespaceName",
+        "AddedBy",
+        "Version",
+        "BaseScores",
+        "CVEIds",
+        "Vulnerabilities",
+    ],
 )
 Layer = namedtuple("Layer", ["Name", "NamespaceName", "ParentName", "IndexedByVersion", "Features"])
 
@@ -64,6 +114,8 @@ def from_dict(cls, data_dict):
                         NamespaceName=f["NamespaceName"],
                         AddedBy=f["AddedBy"],
                         Version=f["Version"],
+                        BaseScores=f["BaseScores"],
+                        CVEIds=f["CVEIds"],
                         Vulnerabilities=[
                             Vulnerability(
                                 Severity=vuln.get("Severity", None),
@@ -111,6 +163,8 @@ def to_dict(self):
                         "NamespaceName": f.NamespaceName,
                         "AddedBy": f.AddedBy,
                         "Version": f.Version,
+                        "BaseScores": vulns_to_base_scores(f.Vulnerabilities),
+                        "CVEIds": vulns_to_cves(f.Vulnerabilities),
                         "Vulnerabilities": [
                             {
                                 "Severity": v.Severity,

data/secscan_model/secscan_v4_model.py

@@ -32,6 +32,7 @@
     SecurityInformationLookupResult,
     UpdatedVulnerability,
     Vulnerability,
+    link_to_cves,
 )
 from data.secscan_model.interface import (
     InvalidConfigurationException,
@@ -605,13 +606,31 @@ def features_for(report):
             else {}
         )
 
+        base_scores = []
+        if report.get("enrichments", {}):
+            for enrichment_list in report["enrichments"].values():
+                for pkg_vuln in enrichment_list:
+                    for k, v in pkg_vuln.items():
+                        if not isinstance(v, list):
+                            logger.error(f"Unexpected type for value of key '{k}': {type(v)}")
+                            continue
+                        for item in v:
+                            if not isinstance(item, dict) or "baseScore" not in item:
+                                logger.error(f"Invalid item format or missing 'baseScore': {item}")
+                                continue
+                            base_scores.append(item["baseScore"])
+
+        cve_ids = [link_to_cves(v["links"]) for v in pkg_vulns]
+
         features.append(
             Feature(
                 pkg["name"],
                 "",
                 "",
                 pkg_env["introduced_in"],
                 pkg["version"],
+                base_scores,
+                cve_ids,
                 [
                     Vulnerability(
                         fetch_vuln_severity(vuln, enrichments),