[v1.3.0] Only trigger vuln threshold on fixable vulns

.github/workflows/test_vuln_thresholds.yml

@@ -30,7 +30,7 @@ jobs:
           role-to-assume: ${{ secrets.AWS_IAM_ROLE }}
 
       - name: Scan artifact with Inspector
-        uses: aws-actions/vulnerability-scan-github-action-for-amazon-inspector@v1
+        uses: aws-actions/vulnerability-scan-github-action-for-amazon-inspector@threshold_fixable_only
         id: inspector
         with:
           artifact_type: 'archive'
@@ -45,6 +45,7 @@ jobs:
           low_threshold: 1
           other_threshold: 1
           sbomgen_version: "latest"
+          threshold_fixable_only: true
 
       - name: Fail if vulnerability threshold is exceeded
         run: if [[  ${{ steps.inspector.outputs.vulnerability_threshold_exceeded }} != "1" ]]; then echo "test failed"; else echo "test passed"; fi

action.yml

@@ -110,6 +110,18 @@ inputs:
     description: "Specifies the OS and CPU arch of the container image you wish to scan. Valid inputs are of the form 'os/cpu/variant' for example, 'linux/amd64', 'linux/arm64/v8', etc. If no platform is specified, the system will use the same platform as the host that is performing the scan. This argument only affects container image scans. Requires inspector-sbomgen 1.5.1 or later."
     required: False
 
+  threshold_fixable_only:
+    description: 'If set to true, only count vulnerabilities with a fix towards threshold exceeded condition.'
+    required: False
+    default: false
+    type: boolean
+
+  show_only_fixable_vulns:
+    description: "If set to true, this action will show only fixed vulnerabilities in the GitHub Actions step summary page. All vulnerability metadata is still retained in the raw Inspector scan files."
+    required: False
+    default: false
+    type: boolean
+
 outputs:
   artifact_sbom:
     description: "The filepath to the artifact's software bill of materials."
@@ -148,6 +160,8 @@ runs:
     - --out-dockerfile-scan-md=${{ inputs.output_inspector_dockerfile_scan_path_markdown }}
     - --sbomgen-version=${{ inputs.sbomgen_version }}
     - --thresholds
+    - ${{ inputs.threshold_fixable_only == 'true' && '--threshold-fixable-only' || '' }}
+    - ${{ inputs.show_only_fixable_vulns == 'true' && '--show-only-fixable-vulns' || '' }}
     - --critical=${{ inputs.critical_threshold }}
     - --high=${{ inputs.high_threshold }}
     - --medium=${{ inputs.medium_threshold }}

entrypoint/entrypoint/cli.py

@@ -1,6 +1,5 @@
 import argparse
 
-
 def init(sys_argv=None) -> argparse.Namespace:
     """
     initializes the CLI using argparse :param: a list of arguments; you should pass sys.argv in most cases.
@@ -50,7 +49,10 @@ def init(sys_argv=None) -> argparse.Namespace:
                         help="Specifies one or more files and/or directories that should NOT be inventoried.")
     parser.add_argument("--timeout", type=str, default="600",
                         help="The amount of time in seconds that inspector-sbomgne will run. When this timeout is exceeded, sbomgen will gracefully conclude and present any findings discovered up to that point.")
-    parser.add_argument("--show-only-fixed-vulnerabilities", action="store_true", help="If set, this program will only show fixed vulnerabilities in the GitHub Actions job summary page.")
+    parser.add_argument("--show-only-fixable-vulns", action="store_true",
+                        help="Only show fixed vulnerabilities in the GitHub Actions job summary page.")
+    parser.add_argument("--threshold-fixable-only", action="store_true",
+                        help="Only count vulnerabilities with a fix towards threshold exceeded condition.")
 
     parser.add_argument("--platform", type=str,
                         help="Specifies the OS and CPU arch of the container image you wish to scan. Valid inputs are "

entrypoint/entrypoint/fixed_vulns.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class FixedVulns:
+    criticals: int
+    highs: int
+    mediums: int
+    lows: int
+    others: int

entrypoint/entrypoint/orchestrator.py

@@ -3,12 +3,13 @@
 import logging
 import os
 import platform
+import re
 import shutil
 import sys
 import tempfile
-import re
+import typing
 
-from entrypoint import dockerfile, executor, exporter, installer, pkg_vuln
+from entrypoint import dockerfile, executor, exporter, installer, pkg_vuln, fixed_vulns
 
 
 def execute(args) -> int:
@@ -26,12 +27,13 @@ def execute(args) -> int:
     set_github_actions_output('inspector_scan_results', args.out_scan)
 
     logging.info("tallying vulnerabilities")
-    succeeded, scan_result = get_scan_result(args)
+    succeeded, scan_result, fixed_vuln_counts = get_scan_result(args)
     require_true(succeeded, "unable to tally vulnerabilities")
 
     print_vuln_count_summary(scan_result)
 
-    set_flag_if_vuln_threshold_exceeded(args, scan_result)
+    vuln_counts = fixed_vuln_counts if args.threshold_fixable_only else scan_result
+    set_env_var_if_vuln_threshold_exceeded(args, vuln_counts)
 
     write_pkg_vuln_report_csv(args.out_scan_csv, scan_result)
     set_github_actions_output('inspector_scan_results_csv', args.out_scan_csv)
@@ -166,7 +168,7 @@ def invoke_sbomgen(args) -> int:
 
     elif "archive" in args.artifact_type.lower():
         args.artifact_type = "archive"
-        path_arg = "--path"        
+        path_arg = "--path"
 
     else:
         logging.error(
@@ -201,7 +203,8 @@ def invoke_sbomgen(args) -> int:
         if args.platform:
             platform_arg = args.platform.lower()
             if not is_valid_container_platform(platform_arg):
-                logging.fatal(f"received invalid container image platform: '{args.platform}'. Platform should be of the form 'os/cpu/variant' such as 'linux/amd64' or 'linux/arm64/v8'")       
+                logging.fatal(
+                    f"received invalid container image platform: '{args.platform}'. Platform should be of the form 'os/cpu/variant' such as 'linux/amd64' or 'linux/arm64/v8'")
             sbomgen_args.append("--platform")
             sbomgen_args.append(platform_arg)
 
@@ -232,40 +235,44 @@ def invoke_inspector_scan(src_sbom, dst_scan):
     return ret
 
 
-def get_scan_result(args) -> tuple[bool, exporter.InspectorScanResult]:
-    if args.show_only_fixed_vulnerabilities == True:
-        succeeded, criticals, highs, mediums, lows, others = get_fixed_vuln_counts(args.out_scan)
-        if succeeded is False:
-            return False, None
-    else:
-        succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
-        if succeeded is False:
-            return False, None
+def get_scan_result(args) -> tuple[bool, exporter.InspectorScanResult, fixed_vulns.FixedVulns]:
+    scan_result = exporter.InspectorScanResult(vulnerabilities=[pkg_vuln.Vulnerability()])
+    fixed_vulns_counts = fixed_vulns.FixedVulns(criticals=0, highs=0, mediums=0, lows=0, others=0)
+
+    succeeded, fixed_vulns_counts = get_fixed_vuln_counts(
+        args.out_scan)
+    if succeeded is False:
+        return False, scan_result, fixed_vulns_counts
+
+    succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
+    if succeeded is False:
+        return False, scan_result, fixed_vulns_counts
+
 
     try:
         with open(args.out_scan, "r") as f:
             inspector_scan = json.load(f)
             vulns = pkg_vuln.parse_inspector_scan_result(inspector_scan)
-            if args.show_only_fixed_vulnerabilities:
-                for vuln in vulns:
-                    if vuln.fixed_ver == "null":
-                        vulns.remove(vuln)
+            #if args.show_only_fixed_vulnerabilities:
+                #for vuln in vulns:
+                    #if vuln.fixed_ver == "null":
+                        #vulns.remove(vuln)
     except Exception as e:
         logging.error(e)
-        return False, None
-    
+        return False, scan_result, fixed_vulns_counts
+
     scan_result = exporter.InspectorScanResult(
         vulnerabilities=vulns,
         artifact_name=args.artifact_path,
         artifact_type=args.artifact_type,
-        criticals=criticals,
-        highs=highs,
-        mediums=mediums,
-        lows=lows,
-        others=others
+        criticals=str(criticals),
+        highs=str(highs),
+        mediums=str(mediums),
+        lows=str(lows),
+        others=str(others)
     )
 
-    return succeeded, scan_result
+    return succeeded, scan_result, fixed_vulns_counts
 
 
 def set_github_actions_output(key, value):
@@ -276,6 +283,10 @@ def set_github_actions_output(key, value):
         logging.info(f"setting github actions output: {key}:{value}")
         os.system(f'echo "{key}={value}" >> "$GITHUB_OUTPUT"')
 
+    # set an ENV VAR so that outputs
+    # look the same locally or on GitHub Actions
+    os.environ[key] = str(value)
+
     return
 
 
@@ -340,78 +351,83 @@ def get_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int,
 
     return True, criticals, highs, mediums, lows, others
 
-def get_fixed_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int, int]:
-    criticals_fixed = 0
-    highs_fixed = 0
-    mediums_fixed = 0
-    lows_fixed = 0
-    others_fixed = 0
+
+def get_fixed_vuln_counts(inspector_scan_path: str) -> tuple[bool, fixed_vulns.FixedVulns]:
+    fixed_vulns_counts = fixed_vulns.FixedVulns(criticals=0, highs=0, mediums=0, lows=0, others=0)
 
     scan_contents = ""
     try:
         with open(inspector_scan_path, 'r') as f:
             scan_contents = json.load(f)
     except Exception as e:
         logging.error(e)
-        return False, criticals, highs, mediums, lows, others
+        return False, fixed_vulns_counts
 
     # find the sbom->metadata->properties object
     scan_contents = scan_contents.get("sbom")
     if scan_contents is None:
         logging.error(
             f"expected Inspector scan results with 'sbom' as root object, but it was not found in file {inspector_scan_path}")
-        return False, criticals, highs, mediums, lows, others
+        return False, fixed_vulns_counts
 
     vulnerabilities = scan_contents.get("vulnerabilities")
 
     if vulnerabilities is None:
         # no vulnerabilities found
-        return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
-
+        return True, fixed_vulns_counts
+
+    is_fix_available_key = "amazon:inspector:sbom_scanner:fixed_version:comp-"
     for vuln in vulnerabilities:
         props = vuln.get("properties")
         if props is None:
-            logging.error(f"expected vulnerability with 'properties' key but none was found in file {inspector_scan_path}")
+            logging.error(
+                f"expected vulnerability with 'properties' key but none was found in file {inspector_scan_path}")
             continue
 
         for prop in props:
             name = prop.get("name")
             if name is None:
-                logging.error(f"expected vulnerability with 'name' key but none was found in file {inspector_scan_path}")
                 continue
-            if "amazon:inspector:sbom_scanner:fixed_version:comp-" in name:
-                rating = vuln.get("ratings")
-                if rating is None:
-                    logging.error(f"expected vulnerability with 'rating' key but none was found in file {inspector_scan_path}")
+            if is_fix_available_key not in name:
+                continue
+
+            # vuln has an available fix
+            rating = vuln.get("ratings")
+            if rating is None:
+                logging.error(
+                    f"expected vulnerability with 'rating' key but none was found in file {inspector_scan_path}")
+                continue
+
+            previous_score = 0
+            previous_severity = ""
+            for each_rating in rating:
+                severity = each_rating.get("severity")
+                if severity is None:
+                    logging.error(
+                        f"expected vulnerability with 'severity' key but none was found in file {inspector_scan_path}")
+                    continue
+                score = each_rating.get("score")
+                if score is None:
+                    logging.debug(
+                        f"expected vulnerability with 'score' key but none was found in file {inspector_scan_path}")
                     continue
+                if previous_score < score:
+                    previous_score = score
+                    previous_severity = severity
+
+            if previous_severity == "critical":
+                fixed_vulns_counts.criticals += 1
+            elif previous_severity == "high":
+                fixed_vulns_counts.highs += 1
+            elif previous_severity == "medium":
+                fixed_vulns_counts.mediums += 1
+            elif previous_severity == "low":
+                fixed_vulns_counts.lows += 1
+            else:
+                fixed_vulns_counts.others += 1
+
+    return True, fixed_vulns_counts
 
-                previous_score = 0
-                previous_severity = ""
-                for each_rating in rating:
-                    severity = each_rating.get("severity")
-                    if severity is None:
-                        logging.error(f"expected vulnerability with 'severity' key but none was found in file {inspector_scan_path}")
-                        continue
-                    score = each_rating.get("score")
-                    if score is None:
-                        logging.error(f"expected vulnerability with 'score' key but none was found in file {inspector_scan_path}")
-                        continue
-                    if previous_score < score:
-                        previous_score = score
-                        previous_severity = severity
-
-                if previous_severity == "none":
-                    others_fixed += 1
-                elif previous_severity == "critical":
-                    criticals_fixed += 1
-                elif previous_severity == "high":
-                    highs_fixed += 1
-                elif previous_severity == "medium":
-                    mediums_fixed += 1
-                elif previous_severity == "low":
-                    lows_fixed += 1
-
-    return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
 
 def install_sbomgen(args):
     os_name = platform.system()
@@ -452,12 +468,14 @@ def write_pkg_vuln_report_markdown(out_scan_markdown, scan_result: exporter.Insp
     return markdown
 
 
-def set_flag_if_vuln_threshold_exceeded(args, scan_result: exporter.InspectorScanResult):
-    is_exceeded = exceeds_threshold(scan_result.criticals, args.critical,
-                                    scan_result.highs, args.high,
-                                    scan_result.mediums, args.medium,
-                                    scan_result.lows, args.low,
-                                    scan_result.others, args.other)
+def set_env_var_if_vuln_threshold_exceeded(args,
+                                           vuln_counts: typing.Union[
+                                               exporter.InspectorScanResult, fixed_vulns.FixedVulns]):
+    is_exceeded = exceeds_threshold(vuln_counts.criticals, args.critical,
+                                    vuln_counts.highs, args.high,
+                                    vuln_counts.mediums, args.medium,
+                                    vuln_counts.lows, args.low,
+                                    vuln_counts.others, args.other)
 
     if is_exceeded and args.thresholds:
         set_github_actions_output('vulnerability_threshold_exceeded', 1)
@@ -471,19 +489,19 @@ def exceeds_threshold(criticals, critical_threshold,
                       lows, low_threshold,
                       others, other_threshold) -> bool:
     is_threshold_exceed = False
-    if 0 < critical_threshold <= criticals:
+    if 0 < critical_threshold <= int(criticals):
         is_threshold_exceed = True
 
-    if 0 < high_threshold <= highs:
+    if 0 < high_threshold <= int(highs):
         is_threshold_exceed = True
 
-    if 0 < medium_threshold <= mediums:
+    if 0 < medium_threshold <= int(mediums):
         is_threshold_exceed = True
 
-    if 0 < low_threshold <= lows:
+    if 0 < low_threshold <= int(lows):
         is_threshold_exceed = True
 
-    if 0 < other_threshold <= others:
+    if 0 < other_threshold <= int(others):
         is_threshold_exceed = True
 
     return is_threshold_exceed
@@ -533,6 +551,7 @@ def require_true(expr: bool, msg: str):
         logging.error(msg)
         exit(1)
 
+
 def is_valid_container_platform(img_platform):
     # regex for detecting 'os/cpu/variant'
     # os/cpu are required whereas variant is optional